RabbitMQ 入门指南（二）

向右看齐 2022-12-16 03:58 231阅读 0赞

## Work Queue ##

Work Queue，又称为 Task Queue，用于将耗时的 task（任务）分发给多个 worker。

对于资源密集型（消耗资源）的任务，我们往往需要等待它处理完毕，才能处理下一个任务。

这样是不合理的，而 Work Queue 可以解决这个问题。

Work Queue 可以计划安排稍后再处理消息任务，它将任务封装压缩为一条消息，存到队列中。后台运行的 worker 进程将会弹出 task，并最终执行对应的 job。

**多个 worker 之间共享所有的 task，注意是平均分摊。同一个 task 只会被一个 worker 消费。**

在 Web 应用中，Work Queue 非常有用，因为短暂的 HTTP 请求无法处理复杂的、耗时的任务。

![image][]

## 准备工作 ##

这里，我们用字符串消息来代表复杂的 task，即假定它是一条耗时的任务。

我们简单修改上一节中的 send.php 文件中的代码，以便我们可以通过命令行发送任意的消息数据，并将其命名为 new\_task.php。

new\_task.php 内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明队列（queue）
    // queue_declare() 方法是幂等的，即只有当指定的 queue 不存在时才会创建
    $channel->queue_declare('task_queue', false, false, false, false);
    
    // 创建消息
    $data = implode(' ', array_slice($argv, 1));
    if (empty($data)) {
        $data = "Hello World!";
    }
    $msg = new AMQPMessage(
        $data,
        array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
    );
    
    // 将消息发送到指定的队列
    $channel->basic_publish($msg, '', 'task_queue');
    
    // 消息发送成功后，输出提示信息
    echo ' [x] Sent ', $data, "\n";
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

上一节中的 receive.php 文件也需简单修改，我们用消息体中的每一个 **.** 号来假装 1 秒钟（3 个点表示需要 3 秒），用 sleep() 方法将其伪造为一个耗时的任务，并将其命名为 worker.php。

worker.php 内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明要监听的队列（queue）
    $channel->queue_declare('task_queue', false, false, false, false);
    
    // 提示信息
    echo " [*] Waiting for messages. To exit press CTRL+C\n";
    
    // 回调函数
    $callback = function ($msg) {
        // todo 处理消息的业务逻辑
        echo ' [x] Received ', $msg->body, "\n";
        sleep(substr_count($msg->body, '.'));
        echo " [x] Done\n";
    };
    
    // 启动消费者
    $channel->basic_consume('task_queue', '', false, true, false, false, $callback);
    
    while ($channel->is_consuming()) { // 如果正在进行回调处理
        $channel->wait(); // 等待
    }
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

**注意：** 这里，我们只是在模拟一个假的耗时任务。

然后，运行它们。

# shell 1，运行 Consumer
    php worker.php
    
    # shell 2，发送消息
    php new_task.php "A busy task which takes two seconds.."

## 轮询分发 ##

Task Queue 的一个显著优点是它可以并行工作，我们只需新增多个 worker 进程，便可以处理积压的 task。

比如，我们同时运行两个 worker.php 脚本。

# shell 1，运行第一个消费者 C1
    php worker.php

# shell 2，运行第二个消费者 C2
    php worker.php

然后，我们打开第三个命令行窗口（shell），来发送多条消息。

# shell 3
    php new_task.php 1th message.
    php new_task.php 2th message..
    php new_task.php 3th message...
    php new_task.php 4th message....
    php new_task.php 5th message.....

下面是 C1 和 C2 的输出结果：

# 消费者 C1
     [*] Waiting for messages. To exit press CTRL+C
     [x] Received 1th message.
     [x] Done
     [x] Received 3th message...
     [x] Done
     [x] Received 5th message.....
     [x] Done

# 消费者 C2
     [*] Waiting for messages. To exit press CTRL+C
     [x] Received 2th message..
     [x] Done
     [x] Received 4th message....
     [x] Done

默认情况下，RabbitMQ 会依次将每条消息发送给下一个 Consumer，所有的 task 消息会被多个 Consumer（worker 进程）平均分摊。这种分配（分发）消息的机制，被称为**轮询**。

## Message acknowledgment ##

**消息回执**（消息确认）是指消费者在消费完消息后手动发送一个回执（ack）给 RabbitMQ，RabbitMQ 收到 ack 回执后，才会把消息从内存中删除（实质上是先打上删除标记，之后再删除）。

处理一条消息可能需要花费几秒甚至更长的时间，如果消费者在处理耗时消息的过程中，仅完成了部分业务逻辑就死掉了，会发生什么情况？

基于当前的代码，一旦 RabbitMQ 发送完一条 message 给 consumer，RabbitMQ 就会立即将该消息标记为删除。这种情况下，如果你杀死了 worker 进程或者 worker 突然意外死掉了，那么该 worker 中正在处理的消息就会丢失。所有分配给该 worker 的未处理完毕的消息就都丢失了。

但事实上，消息都有用，我们不能丢失任何消息。

**为了确保消息不丢失，RabbitMQ 提供了 Message acknowledgment（消息回执）的支持。** 当某条特定的消息被 consumer 接收并处理完毕后，consumer 就会返回一个 **ack** （回执）给 RabbitMQ，这样，RabbitMQ 才会将该消息删除。

开启消息回执功能后，如果某个 consumer 死掉了却没有发送 ack，RabbitMQ 就会认为该消息没有处理完毕，并让消息重新回到队列。如果有其他的 consumer 正在运行，RabbitMQ 就会迅速重发该消息到正常运行的另一个 consumer。这样，即使 consumer 偶尔都死掉了，消息也不会丢失。

只有当 consumer 死掉时，RabbitMQ 才会重发消息。因此不存在超时的问题，即使处理消息的业务逻辑需要花费很长的时间。

**消息回执的功能默认是关闭的，autoAck 参数的值默认为 true。**

RabbitMQ 队列中的消息分为两个部分：一部分是等待发送给消费者的消息；另一部分是已经发送给消费者但还未收到消费者的 ack 信号的消息。

当把 autoAck 参数的值设置为 false，如果 RabbitMQ 一直没有收到消费者的 ack 信号，并且该消费者已断开了连接，那么，RabbitMQ 会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的消费者。

### 开启消息回执功能 ###

1.  将 basic\_consume() 方法的第四个参数设置为 false。
2.  在回调函数 $callback 的内部末尾，发送合适的回执。

为此，需要修改 worker.php 的部分代码。

// 回调函数
    $callback = function ($msg) {
        // todo 处理消息的业务逻辑
        echo ' [x] Received ', $msg->body, "\n";
        sleep(substr_count($msg->body, '.'));
        echo " [x] Done\n";
        // 发送回执
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    };
    
    // 启动消费者
    $channel->basic_consume('task_queue', '', false, false, false, false, $callback);

这样，即使你使用 CTRL+C 终止了正在处理消息的 worker 进程，消息也不会丢失。因为，worker 终止后，很快所有没有回执的消息都会重新发送。

**注意事项：**

*  修改 worker.php 的代码后，需要重新运行 worker.php 脚本，修改才会生效。
 *  发送 Ack（回执）的 channel 必须和接收消息的 channel 相同，否则，会导致 channel-level 的协议异常。

### 忘记发送回执 ###

一个常见的错误是，虽然我们开启了回执功能，却遗漏了发送回执。这会导致严重的后果。

当 consumer 客户端退出时，消息会被重新发送，看似只是消息的随机重发，实则会导致两种后果。

*  没有回执的消息，RabbitMQ 无法释放，就会导致 RabbitMQ 吃掉越来越多的内存。
 *  consumer 终止后，消息会重发，导致消息的重复消费。

为了调试这种错误，可以使用 rabbitmqctl 命令打印 messages\_unacknowledged 字段信息。

rabbitmqctl list_queues name messages_ready messages_unacknowledged

当然，也可以使用 RabbitMQ 的可视化 Web 管理界面，来查看没有回执的消息数。

http://127.0.0.1:15672/#/queues

## Message durability ##

上面我们已经学会了如果 consumer 死掉了，如何确保消息不丢失。但是，如果 RabbitMQ 服务器挂掉了，消息还是会丢失。

默认情况下，如果 RabbitMQ 服务器终止或者宕机了，内部的队列和消息就会丢失。

为了确保 RabbitMQ 宕机时，消息不丢失，需要做两件事情：

1.  标记 queue 持久化
2.  标记 message 持久化

**首先，我们需要确保 RabbitMQ 不会丢失 queue。我们将 queue\_declare() 方法的第三个参数设置为 true 即可。**

$channel->queue_declare('task_queue', false, true, false, false);

**注意事项：**

*  RabbitMQ 不允许用不同的参数重新定义一个已经存在的队列，如果你非要这样做，会返回一个错误。快速的应对方案是声明一个不同名称的队列。
 *  producer 和 consumer 中都应当这样声明队列。

**其次，我们需要确保 message 是持久的。创建消息时，设置消息的属性 delivery\_mode = 2。**

$msg = new AMQPMessage(
        $data,
        array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
    );

**以上两点就是消息的持久化机制，它可以保证绝大部分情况下的消息不丢失。**

虽然**消息持久化**会告知 RabbitMQ 将消息持久化到磁盘，但是它仍然无法完全保证消息不丢失。

比如，有这样一种小概率事件，RabbitMQ 接收到了生产者的消息，还没来得及保存。另外，RabbitMQ 并不会把每一条消息都同步到磁盘，它可能仅仅是保存到缓存，并没有真正地写入磁盘。

如果你想要实现更为健壮的消息安全保障，可以使用 **[publisher confirms][]**。

## Fair dispatch（合理分配） ##

你可能注意到多个 worker 之间的消息分配并没有严格按照我们期望的那样进行。

比如，有 2 个 worker，当所有奇数的消息任务比偶数的消息繁重时，就会出现一个 worker 一直处于繁忙状态，而另一个 worker 几乎处于闲置状态。然而，RabbitMQ 并不知道此事，仍然会均匀地分发消息。显然，这样是不合理的。

这种问题之所以会发生，是因为消息进入队列时，RabbitMQ 只是分配消息，它没有考虑某一个 consumer 中没有回执的消息数。它只是盲目地分配第 n 条 message 到 第 n 个 consumer.

为了解决这个问题，可以使用 **basic\_qos()** 方法，设置 prefetch\_count = 1。它告诉 RabbitMQ 每次只能给一个 consumer 分配一条 message。

也就是说，只有当 consumer 处理完毕并回执了前一条消息，RabbitMQ 才会分配一条新的消息给该 consumer。这样，它就会把消息分配给空闲的 consumer。

在 worker.php 中，添加如下的代码：

$channel->basic_qos(null, 1, null);

这样，就实现了消息的合理分配。

**注意：** 如果所有的 worker 都很繁忙，队列可能会堵塞。你需要随时留意队列中的消息数，并及时添加更多的 worker，或者采用其他的策略。

## 最终的代码示例 ##

new\_task.php 文件，内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明队列（queue）
    // queue_declare() 方法是幂等的，即只有当指定的 queue 不存在时才会创建
    $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' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
    );
    
    // 将消息发送到指定的队列
    $channel->basic_publish($msg, '', 'task_queue');
    
    // 消息发送成功后，输出提示信息
    echo ' [x] Sent ', $data, "\n";
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

worker.php 文件，内容如下：

<?php
    
    // php 的依赖包自动加载，引入相关的类
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    
    // 连接到指定的 RabbitMQ 服务器（Broker），并创建 channel
    $connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    // 声明要监听的队列（queue）
    $channel->queue_declare('task_queue', false, true, false, false);
    
    // 提示信息
    echo " [*] Waiting for messages. To exit press CTRL+C\n";
    
    // 回调函数
    $callback = function ($msg) {
        // todo 处理消息的业务逻辑
        echo ' [x] Received ', $msg->body, "\n";
        sleep(substr_count($msg->body, '.'));
        echo " [x] Done\n";
        // 发送回执
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    };
    
    // 每次只能被分配一条 message
    $channel->basic_qos(null, 1, null);
    
    // 启动消费者
    $channel->basic_consume('task_queue', '', false, false, false, false, $callback);
    
    while ($channel->is_consuming()) { // 如果正在进行回调处理
        $channel->wait(); // 等待
    }
    
    // 最后，关闭 channel 和 connection
    $channel->close();
    $connection->close();

**参考文献**

\[1\] [https://www.rabbitmq.com/tutorials/tutorial-two-php.html][https_www.rabbitmq.com_tutorials_tutorial-two-php.html]

[image]: /images/20221123/831ecc47e5124fd09aed469ce684721a.png
[publisher confirms]: https://www.rabbitmq.com/confirms.html
[https_www.rabbitmq.com_tutorials_tutorial-two-php.html]: https://www.rabbitmq.com/tutorials/tutorial-two-php.html