关于rabbitmq:RabbitMQ-工作队列的使用

工作队列的利用场景通常为一个生产者、一个队列、多个消费者偏心生产队列。

工作队列通常配合以下设置独特应用：

• 开启队列长久化
• 开启音讯长久化
• 开启偏心散发
• 敞开主动音讯确认
• 手动确认音讯

装置依赖

# composer.json
{
    "require": {"php-amqplib/php-amqplib": ">=3.0"}
}

> composer.phar install

模式构造

生产者

生产者连贯到 RabbitMQ，发送一条音讯，而后退出。

# send.php

<?php
require_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
<?php
require_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.....

注意事项

1. 敞开主动音讯确认必须记得要手动确认，否则会导致音讯无奈开释，内存耗费越来越大。
2. RabbitMQ 不容许应用不同的参数从新定义已有队列，将返回谬误，如非长久化的队列设置为长久化的队列等。

查看未确认音讯

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged