如何在RabbitMQ与PHP中逐个执行任务而非并行处理？

要实现任务逐个按顺序执行（比如10个任务完成一个再跑下一个），核心是避免RabbitMQ同时给多个消费者分发消息，并且确保单个消费者一次只处理一条消息。下面是具体的实现思路和代码示例：

核心原理 #

RabbitMQ的队列本身是FIFO（先进先出）的，但默认会把消息轮询分发给多个消费者。要实现顺序执行，我们需要做到两点：

1. 确保同一个时刻只有一个消费者进程在监听目标队列；
2. 让RabbitMQ一次只给这个消费者发送一条消息，直到消费者明确确认任务完成，再投递下一条。

具体实现（基于php-amqplib库） #

1. 生产者代码（发送10个任务） #

首先编写生产者，把10个任务发送到指定队列：

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// 建立RabbitMQ连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明持久化队列（避免重启RabbitMQ丢失消息）
$channel->queue_declare('sequential_tasks', false, true, false, false);

// 生成10个任务并发送
for ($taskId = 1; $taskId <= 10; $taskId++) {
    $taskData = json_encode([
        'task_id' => $taskId,
        'content' => "这是第{$taskId}个任务",
        'created_at' => date('Y-m-d H:i:s')
    ]);
    
    // 发送持久化消息
    $message = new AMQPMessage($taskData, [
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
    ]);
    
    $channel->basic_publish($message, '', 'sequential_tasks');
    echo "已发送任务 {$taskId}\n";
}

// 关闭连接
$channel->close();
$connection->close();

2. 消费者代码（顺序处理任务） #

消费者是关键，这里要配置两个核心参数：

• basic_qos设置prefetch_count=1：限制RabbitMQ一次只发一条未确认的消息给这个消费者；
• 手动发送ACK：任务处理完成后再告诉RabbitMQ可以投递下一条。

<?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('sequential_tasks', false, true, false, false);

echo "等待任务执行，按Ctrl+C退出\n";

// 任务处理回调
$taskHandler = function (AMQPMessage $msg) {
    $task = json_decode($msg->body, true);
    echo "开始处理任务 {$task['task_id']}: {$task['content']}\n";
    
    // 模拟任务处理耗时（比如2秒）
    sleep(2);
    
    echo "✅ 任务 {$task['task_id']} 处理完成\n";
    
    // 手动发送ACK，告诉RabbitMQ可以投递下一条消息
    $msg->ack();
};

// 关键配置：一次只接收一条未确认的消息
$channel->basic_qos(null, 1, null);

// 启动消费，关闭自动ACK（no_ack=false）
$channel->basic_consume('sequential_tasks', '', false, false, false, false, $taskHandler);

// 持续监听队列
while ($channel->is_open()) {
    $channel->wait();
}

$channel->close();
$connection->close();

关键注意事项 #

• 单消费者实例：必须确保只启动一个消费者进程，如果启动多个，每个消费者都会拿到自己的消息，导致并发执行；
• 幂等性设计：如果消费者进程意外崩溃，未ACK的消息会被RabbitMQ重新投递，所以要保证任务重复执行不会产生副作用；
• 框架场景（比如Laravel）：如果用Laravel的RabbitMQ队列驱动，只需启动单个worker进程即可：

  php artisan queue:work rabbitmq --queue=sequential_tasks --max-processes=1 --sleep=3 --tries=3
  

  --max-processes=1确保只有一个进程处理队列，Laravel会自动处理消息顺序和ACK逻辑。

内容的提问来源于stack exchange，提问作者Kaker