Tag: Yii2 Queue

RabbitMQ Exchange

记录一下使用 Yii2 Queue 插件的 AMQP Interop 驱动连接 RabbitMQ 的时候遇到的一个问题。

配置如下：

'queue1' => [
    'class' => 'yii\queue\amqp_interop\Queue',
    'port' => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'queueName' => 'queue@1',
    'driver' => 'enqueue/amqp-lib',
    'dsn' => 'amqp:',
],
'queue2' => [
    'class' => 'yii\queue\amqp_interop\Queue',
    'port' => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'queueName' => 'queue@2',
    'driver' => 'enqueue/amqp-lib',
    'dsn' => 'amqp:',
],

代码如下：

// 往队列 queue@1 推送一条消息
Yii::$app->queue1->push(new NoticeJob([
    'type' => 'notice',
    'content' => 'RabbitMQ Test',
]));

然而 queue@1 和 queue@2 两个队列同时都收到了消息。

一下子想不到问题到底出在哪里，不妨回想一下 RabbitMQ 发送消息的流程。

Publish path from publisher to consumer via exchange and queue
AMQP 0-9-1 Model

生产者发布的消息并不会直接推送到消费者订阅的 Queue，而是会先发送到 Exchange，然后根据一定的 bindings 规则转发到不同的 Queue。

于是我们再来看这个问题：

Exchange: exchange

由于没有配置 exchangeName，Yii 2 Queue 插件自动把 queue@1 和 queue@2 两个队列都绑定到默认的 Exchange: exchange。

只要修改一下配置文件，这个问题就解决了。

'queue1' => [
    'class' => 'yii\queue\amqp_interop\Queue',
    'port' => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'exchangeName' => 'exchange@1',
    'queueName' => 'queue@1',
    'driver' => 'enqueue/amqp-lib',
    'dsn' => 'amqp:',
],
'queue2' => [
    'class' => 'yii\queue\amqp_interop\Queue',
    'port' => 5672,
    'user' => 'guest',
    'password' => 'guest',
    'exchangeName' => 'exchange@2',
    'queueName' => 'queue@2',
    'driver' => 'enqueue/amqp-lib',
    'dsn' => 'amqp:',
],
Exchange: exchange@1
Exchange: exchange@2

参考：

https://github.com/yiisoft/yii2-queue
https://github.com/yiisoft/yii2-queue/blob/master/docs/guide/driver-amqp-interop.md
https://www.rabbitmq.com/tutorials/amqp-concepts.html

 756 total views