Laravel订阅服务捕获RabbitMQ Fanout调用的最佳实现方式咨询

我之前做过好几个多服务基于RabbitMQ Fanout的消息同步场景，给你一套落地性很强的方案，核心是用自定义Artisan命令+进程管理工具来实现长运行的消费者，不用纠结那些文档不全的第三方包~

一、选对依赖包 #

我推荐用 vladimir-yuldashev/laravel-amqp，这个包是Laravel生态里比较成熟的RabbitMQ封装，能快速帮你处理连接、交换器、队列的声明，不用自己写太多底层代码。安装命令：

composer require vladimir-yuldashev/laravel-amqp

安装后发布配置文件：

php artisan vendor:publish --provider="VladimirYuldashev\LaravelQueueRabbitMQ\LaravelQueueRabbitMQServiceProvider"

然后在.env里配置RabbitMQ连接信息：

RABBITMQ_HOST=你的RabbitMQ地址
RABBITMQ_PORT=5672
RABBITMQ_VHOST=/
RABBITMQ_LOGIN=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_QUEUE=default

二、声明Fanout交换器与绑定队列 #

Fanout交换器的核心是广播消息到所有绑定队列，所以每个服务（服务2、3、4...）都需要声明自己的专属队列，并绑定到同一个Fanout交换器上。

你可以在服务启动时自动完成这个操作，比如写个初始化逻辑或者在消费者命令里做：

// 以服务2为例，其他服务只需修改队列名
use VladimirYuldashev\LaravelQueueRabbitMQ\Facades\RabbitMQ;

// 声明持久化的Fanout交换器
RabbitMQ::exchange()->declare('user_updates_exchange', 'fanout', false, true, false);

// 声明服务2专属的持久化队列
$queueName = 'service2_user_updates_queue';
RabbitMQ::queue()->declare($queueName, false, true, false, false);

// 把队列绑定到Fanout交换器
RabbitMQ::queue()->bind($queueName, 'user_updates_exchange');

三、编写长运行的消费者命令 #

Laravel的自定义Artisan命令天生适合做长运行的守护进程，我们可以创建一个专门的命令来持续监听队列。

先创建命令：

php artisan make:command ConsumeUserUpdates

然后编辑app/Console/Commands/ConsumeUserUpdates.php，把消费逻辑和业务代码结合起来：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use VladimirYuldashev\LaravelQueueRabbitMQ\Facades\RabbitMQ;

class ConsumeUserUpdates extends Command
{
    protected $signature = 'rabbitmq:consume-user-updates';
    protected $description = '持续监听RabbitMQ的UserUpdated消息';

    public function handle()
    {
        // 先确保交换器和队列已绑定（防止服务重启后绑定丢失）
        $exchangeName = 'user_updates_exchange';
        $queueName = 'service2_user_updates_queue'; // 每个服务这里换成自己的队列名

        RabbitMQ::exchange()->declare($exchangeName, 'fanout', false, true, false);
        RabbitMQ::queue()->declare($queueName, false, true, false, false);
        RabbitMQ::queue()->bind($queueName, $exchangeName);

        $this->info('开始监听UserUpdated消息...');

        // 开启阻塞式消费，持续监听队列
        RabbitMQ::channel()->basic_consume(
            $queueName,
            '',
            false,
            false,
            false,
            false,
            function ($message) {
                try {
                    // 解析服务1发送的JSON消息
                    $payload = json_decode($message->body, true);
                    $userUuid = $payload['user_uuid'];

                    // 执行当前服务的业务逻辑（这里写你的代码）
                    $this->processUserUpdate($userUuid);

                    // 确认消息已处理，RabbitMQ会移除该消息
                    $message->ack();
                    $this->info("处理完成用户UUID: {$userUuid}");
                } catch (\Exception $e) {
                    // 异常处理：记录日志，避免进程崩溃
                    $this->error("处理失败: {$e->getMessage()}");
                    // 拒绝消息，false表示不重新入队（根据业务调整）
                    $message->nack(false, false);
                }
            }
        );

        // 保持进程运行，持续监听消息
        while (count(RabbitMQ::channel()->callbacks)) {
            RabbitMQ::channel()->wait();
        }
    }

    /**
     * 当前服务的业务逻辑处理方法
     */
    private function processUserUpdate(string $userUuid)
    {
        // 示例逻辑：
        // 1. 根据UUID拉取最新用户信息
        // 2. 更新本地数据库或缓存
        // 3. 触发内部业务事件
    }
}

四、把消费者做成守护进程 #

直接用php artisan rabbitmq:consume-user-updates运行的话，终端关闭进程就停了，所以需要用进程管理工具来守护它，推荐用supervisor（跨系统通用）或者systemd。

Supervisor配置示例： #

1. 安装supervisor（Ubuntu系统：sudo apt-get install supervisor）
2. 创建配置文件/etc/supervisor/conf.d/laravel-rabbitmq-consumer.conf：

[program:laravel-rabbitmq-consumer]
process_name=%(program_name)s_%(process_num)02d
command=php /你的Laravel项目路径/artisan rabbitmq:consume-user-updates
autostart=true
autorestart=true
user=www-data
numprocs=1
redirect_stderr=true
stdout_logfile=/你的Laravel项目路径/storage/logs/rabbitmq-consumer.log

3. 更新配置并启动：

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start laravel-rabbitmq-consumer:*

这样消费者就会在后台持续运行，即使进程意外退出也会自动重启。

五、服务1发送消息的示例 #

确保服务1把消息发送到Fanout交换器，而不是直接发送到队列：

use VladimirYuldashev\LaravelQueueRabbitMQ\Facades\RabbitMQ;

// 构造消息内容
$message = json_encode([
    'user_uuid' => '目标用户UUID',
    // 其他需要传递的字段
]);

// 发送到Fanout交换器，设置持久化防止消息丢失
RabbitMQ::publish($message, 'user_updates_exchange', '', [
    'content_type' => 'application/json',
    'delivery_mode' => 2, // 持久化消息
]);

关键注意事项 #

• 消息持久化：声明交换器和队列时要设置durable=true，发送消息时设置delivery_mode=2，避免RabbitMQ重启后丢失消息。
• 消息确认：一定要调用$message->ack()确认消息处理完成，否则RabbitMQ会认为消息未处理，重启后会重新发送。
• 异常隔离：消费逻辑里必须捕获异常，防止单个消息处理失败导致整个消费者进程崩溃。
• 队列隔离：每个服务用自己的专属队列，这样各个服务的消费逻辑互不影响，也方便排查问题。

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