最初のチュートリアルでは、名前付きキューからメッセージを送受信するプログラムを作成しました。このチュートリアルでは、時間のかかるメッセージを分散するために使用される ワーク キュー を作成します。 複数のワーカー間のタスク。
前回は、指定したキューに対してメッセージを送受信するプログラムを書きました。今回はタスクキューを作成し、それを使って時間のかかるタスクを複数の「ワーカー」に分散してみます。
ワークキュー (別名: タスクキュー) の背後にある主なアイデアは、リソースを大量に消費するタスクをすぐに実行し、完了するまで待機する必要がないようにすることです。代わりに、タスクを後で実行するようにスケジュールします。メッセージを送ってください バックグラウンドで実行されているワーカー プロセスがタスクをポップし、多くのワーカーを実行すると、タスクがワーカー間で共有されます。 ワーク キュー (タスク キューとも呼ばれます) の主な考え方は、リソースを大量に消費し、完了までに時間がかかるタスクを
すぐに処理することを回避することです。代わりに、このタスクは後で完了するようにスケジュール設定します。タスクをメッセージにカプセル化し、キューに送信します。バックグラウンドで実行されている「ワーカー」プロセスがこれらのタスクを実行し、最終的にタスクを完了します。複数の「ワーカー」を実行すると、それらの間でタスクが共有されます。 この概念は、短い HTTP リクエスト ウィンドウ中に複雑なタスクを処理することが不可能な Web アプリケーションで特に役立ちます。
短い HTTP リクエストで複雑なタスクを処理することは不可能であるという概念は、Web アプリケーションで特に役立ちます。
準備
前回述べたように、「Hello World」メッセージを送信しました。 今回は、複雑なタスクを表すためにいくつかの文字列が送信されます。画像のサイズ変更や PDF のレンダリングなどの現実のタスクがないため、sleep() 関数を使用してタスクに時間がかかるように見せかけ、現実のシーンをシミュレートします
前の例の
send.phpコードを少し変更して、コマンド ラインから任意のメッセージを送信できるようにします。このプログラムはタスクをワーク キューにスケジュールするため、new_task.php: という名前を付けます。 「send.php」を思い出してください。コマンドを通じて任意のメッセージを送信できるように少し変更してみましょう。 このプログラムはタスクをワークキューにスケジュールします
したがって、それを new_task.php という名前にします。
リーリー古い
receive.phpスクリプトにもいくつかの変更が必要です。メッセージ本文のドットごとに 1 秒間の作業を行う必要があります。キューからメッセージをポップしてタスクを実行するので、worker.php: と呼びます。 以前の accept.php も調整する必要があります。作業時間をシミュレートするには、メッセージ内の各ピリオドを 1 秒として扱う必要があります。キューからタスクを取得して実行するので、worker.php と呼びます。
リーリー偽のタスクは実行時間をシミュレートしていることに注意してください。
仮想タスクにはシミュレートされた実行時間が必要であることに注意してください。
チュートリアル 1 と同じように実行します。実行方法は前回と同じです (下記)
リーリーラウンドロビン派遣
まず、2 つの worker.php スクリプトを同時に実行してみましょう。両方ともキューからメッセージを取得しますが、具体的にはどうなるか見てみましょう。
まず、同僚が 2 つの worker.php を実行します。それらはすべてキューからメッセージを取得しますが、Meow はどうなるでしょうか? 読む。
3 つのコンソールを開く必要があります。2 つは、worker.php スクリプトを実行します。これらのコンソールは、C1 と C2 です。
3 つのコンソールを開きます。そのうちの 2 つは、worker.php を実行します。これら 2 つのコンソールは 2 つのコンシューマであり、C1 と C2 と呼びます。リーリー リーリー
3 つ目では、新しいタスクを公開します。コンシューマーを開始したら、いくつかのメッセージを公開できます。3 つ目はタスクを公開するために使用されます。すでにコンシューマを実行している場合は、いくつかのメッセージをパブリッシュします。
リーリー従業員に何が提供されるか見てみましょう:
瞅瞅都啥发给我们的worker了。
shell1$ php worker.php [*] Waiting for messages. To exit press CTRL+C [x] Received 'First message.' [x] Received 'Third message...' [x] Received 'Fifth message.....'
shell2$ php worker.php [*] Waiting for messages. To exit press CTRL+C [x] Received 'Second message..' [x] Received 'Fourth message....'
By default, RabbitMQ will send each message to the next consumer, in sequence. On average every consumer will get the same number of messages. This way of distributing messages is called round-robin. Try this out with three or more workers.
默认嘞，RabbitMQ 会按顺序发送每一条消息到下一个消费者，通常每个消费者会收到相同数量的消息。这种分发消息的方式就叫“round-robin”。试试三个或更多的woker.
Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the customer it immediately removes it from memory. In this case, if you kill a worker we will lose the message it was just processing. We'll also lose all the messages that were dispatched to this particular worker but were not yet handled.
处理一个任务会要好几秒。你可能会想要是其中一个消费者执行一个耗时的任务且死在半路上咋整。按我们现在的代码来说，
一旦RabbitMQ派送一条信息到一个消费者，就会从内存中立即将它删除。这种场景下，如果你K掉一个worker的进程，那么它
正在处理的那条消息就会丢失。同样所有派送到这个worker的消息，在未处理前都会丢失。
But we don't want to lose any tasks. If a worker dies, we'd like the task to be delivered to another worker.
我们肯定不想丢掉任何任务呀。如果一个worker挂了，我们很希望这个任务被分配到其他的worker头上。
In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back from the consumer to tell RabbitMQ that a particular message has been received, processed and that RabbitMQ is free to delete it.
为了保证消息决不丢失，RabbitMQ支持消息确认。啥意思呢？就是说消费者发送一条消息给RabbitMQ告诉它说，某条特定的
消息已经收到且处理了，这样RabbitMQ就可以随便删除它啦。
If a consumer dies without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.
如果消费者没法送（消息）确认就挂了，RabbitMQ就会认为消息没有被完全处理，然后重新派送它到另一个消费者。那样
的话你就能保证不会丢失消息啦，即使worker间歇性死掉都木事。
There aren't any message timeouts; RabbitMQ will redeliver the message only when the worker connection dies. It's fine even if processing a message takes a very, very long time.
当且仅当worker连接挂掉的时候，RabbitMQ才会重新派送消息。所以不存在消息处理超时问题。即使要花好久好久好久好久
....处理一条消息都木事。
Message acknowledgments are turned off by default. It's time to turn them on by setting the fourth parameter to basic_consume to false (true means no ack) and send a proper acknowledgment from the worker, once we're done with a task.
默认情况下，消息确认是关闭的。是时候开启他们了，一旦一项任务完成，通过设置basic_consume 第四个参数设置为false(true意思是关闭消息确认)来发送适当的确认信息。
$callback = function($msg){
echo " [x] Received ", $msg->body, "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);
Using this code we can be sure that even if you kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered.
用这段代码我们就能保证神马都不会丢失，即使你在worker处理消息的时候用CTRL+C把它K掉。要是worker被干掉，不久
所有未确认的消息会被重新派送。
Forgotten acknowledgment（老年失忆）
It's a common mistake to miss the basic_ack. It's an easy error, but the consequences are serious. Messages will be redelivered when your client quits (which may look like random redelivery), but RabbitMQ will eat more and more memory as it won't be able to release any unacked messages.
丢失basic_ack很常见。是一个很容犯的错误，可后果很严重！！！
当你的客户端退出后，消息会被重新派送（看起来就像随机重派一样），RabbitMQ会吃掉你越来越多的内存，
因为它会不释放任何未确认的消息哇！！！
In order to debug this kind of mistake you can use rabbitmqctl to print the messages_unacknowledged field:
要调试这种错误，你可以用rabbitmqctl来打印messages_unacknowledged 字段。
$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged Listing queues ... hello 0 0 ...done.ログイン後にコピー
We have learned how to make sure that even if the consumer dies, the task isn't lost. But our tasks will still be lost if RabbitMQ server stops.
我们已经学了怎么保证消费者挂掉的时候任务不丢失。但是如果RabbitMQ服务挂了，我们的任务仍然会丢失。
When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren't lost: we need to mark both the queue and messages as durable.
当RabbitMQ退出或者挂掉，它会忘掉其中的队列和消息，除非你告诉它表酱紫！！！
要保证消息不丢失，需要做两件事：我们需要把队列和消息都标记为持久的。
First, we need to make sure that RabbitMQ will never lose our queue. In order to do so, we need to declare it as durable. To do so we pass the third parameter to queue_declare as true:
首先，我们需要保证RabbitMQ永远不会丢失我们的队列。为了这么干，我们得声明它为持久的。也就是把queue_declare
的第三个参数设置为true.
$channel->queue_declare('hello', false, true, false, false);
Although this command is correct by itself, it won't work in our present setup. That's because we've already defined a queue called hello which is not durable. RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that. But there is a quick workaround - let's declare a queue with different name, for example task_queue:
尽管这么用是没错的，但当前的设置不会起作用。因为我们已经定义了一个叫hello的非持久的队列。RabbitMQ不允许用不同的参数重新定义一个已存在的队列，并返回错误给任何试图那么做的程序。不过这有一个快捷的解决方案-我们可以声明一个不同名字的队列，例如task_queue:
$channel->queue_declare('task_queue', false, true, false, false);
This flag set to true needs to be applied to both the producer and consumer code.
生产者和消费者的代码都需要将队列声明这里设置为true（第三个参数）。
At this point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by setting the delivery_mode = 2 message property which AMQPMessage takes as part of the property array.
到这里，我们就能保证及时RabbitMQ重启task_queue这个队列也不会丢失。现在呢，我们也得把消息标记为持久的，设置一个数组的属性delivery_mode=2，作为AMQPMessage 的参数（第二个）就可以啦。
$msg = new AMQPMessage($data,
array('delivery_mode' => 2) # make message persistent
);
Note on message persistence（需要注意滴）
Marking messages as persistent doesn't fully guarantee that a message won't be lost. Although it tells RabbitMQ to save the message to disk, there is still a short time window when RabbitMQ has accepted a message and hasn't saved it yet. Also, RabbitMQ doesn't do fsync(2) for every message -- it may be just saved to cache and not really written to the disk. The persistence guarantees aren't strong, but it's more than enough for our simple task queue. If you need a stronger guarantee you can wrap the publishing code in a transaction.
把消息设置为持久的并不能完全保证它不会丢失。尽快它也告诉RabbitMQ把消息保存到硬盘上了，但还是有
很小的一种可能性存在——就是RabbitMQ收到消息但还没来得及保存到硬盘，还有，RabbitMQ不是为每条消息都做fsync操作——很可能只是保存到缓存中而不是真的写到硬盘。这种持久性的保证并不够坚实，但是对我们
简单的任务队列来讲是绰绰有余了。如果你需要更坚实的持久性保证，可以把发送代码包装在一个事务当中。
You might have noticed that the dispatching still doesn't work exactly as we want. For example in a situation with two workers, when all odd messages are heavy and even messages are light, one worker will be constantly busy and the other one will do hardly any work. Well, RabbitMQ doesn't know anything about that and will still dispatch messages evenly.
你可能发现了这个分发工作并没按照我么预想的运行。比如一种情况有两个worker，当所有奇数的消息都非常耗时，就算是不耗时，一个worker会一直繁忙，而另一个几乎不干活。好喵，RabbitMQ对于这种情况毫无所知，还会继续均匀的分发消息。
This happens because RabbitMQ just dispatches a message when the message enters the queue. It doesn't look at the number of unacknowledged messages for a consumer. It just blindly dispatches every n-th message to the n-th consumer.
发生这种情况的原因是当信息进入到队列RabbitMQ只是进行分发。它不会去看某个消费者未确认的消息。它只会盲目地分发第N个消息到第N个消费者。
In order to defeat that we can use the basic_qos method with the prefetch_count = 1 setting. This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy.
为了解决这种情况，我们可以用basic_qus方法，设置prefetch_count=1. 这会告诉RabbitMQ一次只给一个worker一个消息。
或者，换句话说，在一个worker处理完和确认上一个消息前，不要给他派送新消息。相反，RabbitMQ会把消息派送给下一个
闲置的worker.
$channel->basic_qos(null, 1, null);
Note about queue size(注意啦！！队列大小)
If all the workers are busy, your queue can fill up. You will want to keep an eye on that, and maybe add more workers, or have some other strategy.
如果所有worker都处于繁忙状态，你的队列是会被填满的。要留心一下，要么增加更多的worker，要么有一些其他的策略。
Final code of our new_task.php file:（new_task.php的终极神码）
channel();
$channel->queue_declare('task_queue', false, true, false, false);
$data = implode(' ', array_slice($argv, 1));
if(empty($data)) $data = "Hello World!";
$msg = new AMQPMessage($data,
array('delivery_mode' => 2) # make message persistent
);
$channel->basic_publish($msg, '', 'task_queue');
echo " [x] Sent ", $data, "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
$channel->close();
$connection->close();
?>
(new_task.php source)
And our worker.php: （worker哦）
channel();
$channel->queue_declare('task_queue', false, true, false, false);
echo ' [*] Waiting for messages. To exit press CTRL+C', "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
$callback = function($msg){
echo " [x] Received ", $msg->body, "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "翻訳-PHP RabbitMQ チュートリアル-2_PHP チュートリアルn";
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_qos(null, 1, null);
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);
while(count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
?>
(worker.php source)
Using message acknowledgments and prefetch you can set up a work queue. The durability options let the tasks survive even if RabbitMQ is restarted.
你可以使用消息确认和预抓取来创建一个工作队列。持久性设置使得即使RabbitMQ重启了任务还会存活。
Now we can move on to tutorial 3 and learn how to deliver the same message to many consumers.
下一扒，我们会学习咋给多个消费者发送相同的消息，就像广播，嗯嗯！
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
