In the first tutorial we wrote programs to send and receive messages from a named queue. In this one we'll create a Work Queue that will be used to distribute time-consuming tasks among multiple workers.
Last time, we wrote a program to send and receive messages from a specified queue. This time, we are going to create a task queue and use it to distribute time-consuming tasks to multiple "workers".
The main idea behind Work Queues (aka: Task Queues) is to avoid doing a resource-intensive task immediately and having to wait for it to complete. Instead we schedule the task to be done later. We encapsulate a task as a message and send it to a queue. A worker process running in the background will pop the tasks and eventually execute the job. When you run many workers the tasks will be shared between them.
The main idea of a work queue (also called a task queue) is to avoid immediately processing tasks that are resource-intensive and take a long time to complete. Instead, we schedule this task to be completed later. We encapsulate tasks into messages and send them to a queue. A "worker" process running in the background will take out these tasks and eventually complete the task. When you run multiple "workers", tasks are shared among them.
This concept is especially useful in web applications where it's impossible to handle a complex task during a short HTTP request window.
The concept that it is impossible to handle a complex task in a short HTTP request is particularly useful in web applications.
In the previous part of this tutorial we sent a message containing "Hello World!". Now we'll be sending strings that stand for complex tasks. We don't have a real-world task, like images to be resized or pdf files to be rendered, so let's fake it by just pretending we're busy - by using the sleep() function. We'll take the number of dots in the string as its complexity; every dot will account for one second of "work". For example, a fake task described by Hello. .. will take three seconds.
As mentioned last time, we sent a "Hello World" message. This time some strings are sent to represent complex tasks. We don’t have real-life tasks like resizing images or rendering pdf, so we use the sleep() function to pretend that the task is time-consuming, and then simulate real-life scenes
We will slightly modify the send.php code from our previous example, to allow arbitrary messages to be sent from the command line. This program will schedule tasks to our work queue, so let's name it new_task.php:
Remember "send.php", let's modify it slightly so that it can send arbitrary messages through commands. This program will schedule tasks into our work queue,
So just call it new_task.php.
$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, "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
Our old receive.php script also requires some changes: it needs to fake a second of work for every dot in the message body. It will pop messages from the queue and perform the task, so let's call it worker .php:
The previous receive.php also needs to be adjusted: each period in the message needs to be treated as one second to simulate the work time. It will pull the task from the queue and execute it, so let's call it worker.php.
$callback = function($msg){
echo " [x] Received ", $msg->body, "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
$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);
Note that our fake task simulates execution time.
Note that our hypothetical task requires simulated execution time.
Run them as in tutorial one: The execution method is the same as last time (below)
shell1$ php new_task.php "A very hard task which takes two seconds.." shell2$ php worker.php
One of the advantages of using a Task Queue is the ability to easily parallelise work. If we are building up a backlog of work, we can just add more workers and that way, scale easily.
The ability to easily work in parallel is one of the advantages of using task queues. If we have a bunch of tasks, just add more workers, it's easy to measure.
First, let's try to run two worker.php scripts at the same time. They will both get messages from the queue, but how exactly? Let's see.
First, our colleagues run two worker.php. They will all get messages from the queue, but what about Meow? Read on.
You need three consoles open. Two will run the worker.php script. These consoles will be our two consumers - C1 and C2.
Open three consoles. Two of them run worker.php. These two consoles are our two consumers, let's call them C1 and C2.
shell1$ php worker.php [*] Waiting for messages. To exit press CTRL+C
shell2$ php worker.php [*] Waiting for messages. To exit press CTRL+C
In the third one we'll publish new tasks. Once you've started the consumers you can publish a few messages:
The third one is used by us to publish tasks. If you already have the consumer(s) running, publish some messages.
shell3$ php new_task.php First message. shell3$ php new_task.php Second message.. shell3$ php new_task.php Third message... shell3$ php new_task.php Fourth message.... shell3$ php new_task.php Fifth message.....
Let's see what is delivered to our workers:
瞅瞅都啥发给我们的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, "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
$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.Copy after login
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, "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
$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', "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
$callback = function($msg){
echo " [x] Received ", $msg->body, "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
sleep(substr_count($msg->body, '.'));
echo " [x] Done", "Translation-PHP rabbitMQ Tutorial-2_PHP Tutorialn";
$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 front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
