工作队列的利用场景通常为一个生产者、一个队列、多个消费者偏心生产队列。
工作队列通常配合以下设置独特应用：
- 开启队列长久化
- 开启音讯长久化
- 开启偏心散发
- 敞开主动音讯确认
- 手动确认音讯
装置依赖
# composer.json{ "require": { "php-amqplib/php-amqplib": ">=3.0" }}> composer.phar install模式构造
生产者
生产者连贯到RabbitMQ，发送一条音讯，而后退出。
# send.php<?phprequire_once __DIR__ . '/vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;use PhpAmqpLib\Message\AMQPMessage;// 创立连贯$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');// 创立通道$channel = $connection->channel();// 创立队列，已存在的不会反复创立，第三个参数为开启队列长久化$channel->queue_declare('task_queue', false, true, false, false);$data = implode(' ', array_slice($argv, 1));if (empty($data)) { $data = "Hello World!";}// 第二个参数 delivery_mode = AMQPMessage::DELIVERY_MODE_PERSISTENT 为设置音讯长久化$msg = new AMQPMessage( $data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));// 通过默认的交换机发送音讯到队列 (音讯内容, 默认交换机, 路由键);$channel->basic_publish($msg, '', 'task_queue');echo ' [x] Sent ', $data, "\n";$channel->close();$connection->close();消费者
消费者监听来自RabbitMQ的音讯，通常须要始终放弃运行状态以监听音讯。
# receive.php<?phprequire_once __DIR__ . '/vendor/autoload.php';use PhpAmqpLib\Connection\AMQPStreamConnection;// 创立连贯$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');// 创立通道$channel = $connection->channel();// 创立队列，已存在的不会反复创立，第三个参数为开启队列长久化$channel->queue_declare('task_queue', false, true, false, false);echo " [*] Waiting for messages. To exit press CTRL+C\n";// 定义回调函数$callback = function ($msg) { echo ' [x] Received ', $msg->body, "\n"; sleep(substr_count($msg->body, '.')); echo " [x] Done\n"; // 手动音讯确认 $msg->ack();};// 设置 prefetch_count = 1，开启偏心散发（默认为循环散发）// 在解决并确认上一条音讯之前，不要将新音讯发送给消费者，而发送给其余消费者$channel->basic_qos(null, 1, null);// 第四个参数设为false敞开主动音讯确认，为true关上主动音讯确认即投递音讯后立即标记为删除$channel->basic_consume('task_queue', '', false, false, false, false, $callback);while ($channel->is_open()) { $channel->wait();}$channel->close();$connection->close();运行
关上一个终端，运行消费者：
php receive.php# => [*] Waiting for messages. To exit press CTRL+C# => [x] Received 'First message.'# => [x] Received 'Third message...'# => [x] Received 'Fifth message.....'关上另一个终端，运行消费者：
php receive.php# => [*] Waiting for messages. To exit press CTRL+C# => [x] Received 'Second message..'# => [x] Received 'Fourth message....'关上另一个终端，运行生产者：
php send.php First message.php send.php Second message..php send.php Third message...php send.php Fourth message....php send.php Fifth message.....注意事项
- 敞开主动音讯确认必须记得要手动确认，否则会导致音讯无奈开释，内存耗费越来越大。
- RabbitMQ 不容许应用不同的参数从新定义已有队列，将返回谬误，如非长久化的队列设置为长久化的队列等。
查看未确认音讯
sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged