RabbitMQ-为什么错误的订户会收到已发布的消息?

本文介绍了RabbitMQ-为什么错误的订户会收到已发布的消息?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我有两项服务，经理和收藏家.

管理器已通过routingKey user.collected订阅了队列COLLECTED_USER，并调用了UserCollected处理程序.
收集器已通过routingKey user.collect订阅了队列COLLECT_USER，并调用了CollectUser处理程序.

Manager is subscribed to Queue COLLECTED_USER with routingKey user.collected and invokes a UserCollected handler.
Collector is subscribed to Queue COLLECT_USER with routingKey user.collect and invokes a CollectUser handler.

可以有多个收集器，因此我已将exclusive设置为false(有关代码，请参见下文).

There can be multiple collectors so I have set exclusive to false (see below for code).

还有其他服务可以监听事件，例如

There are also other services that listen for events like

user.created，
user.updated，
user.deleted

user.created,
user.updated,
user.deleted

此外，还有一些服务可以监听更常见的事件，例如

In addition there are services that listen for more general events like

#.created
user.#

#.created
user.#

以此类推.

所以我正在使用topic交换器.

So I am using a topic exchange.

| exchange | type  | routingKey     | queueName      |
| -------- | ----- | -------------- | -------------  |
| MY_APP   | topic | user.collect   | COLLECT_USER   |
| MY_APP   | topic | user.collected | COLLECTED_USER |

会发生什么:

管理器发布带有routingKey user.collect
收集器获取user.collect消息并调用CollectUser处理程序
收集器的 CollectUser处理程序确实起作用，然后使用routingKey user.collected
管理器获取user.collected消息并调用UserCollected处理程序

Manager publishes message with routingKey user.collect
Collector gets the user.collect message and invokes a CollectUser handler
Collector's CollectUser handler does work, then publishes a message with routingKey user.collected
Manager gets the user.collected message and invokes the UserCollected handler

实际发生的事情:

管理器发布带有routingKey user.collect的消息(正确)
收集器获取user.collect消息并调用CollectUser处理程序(正确)
管理器还会获取user.collect消息，并使用错误的数据调用UserCollected处理程序. (错误)
收集器的 CollectUser处理程序确实起作用，然后发布带有routingKey user.collected(正确)的消息
管理器获取user.collected消息并调用UserCollected处理程序(正确)

Manager publishes message with routingKey user.collect (correct)
Collector gets the user.collect message and invokes a CollectUser handler (correct)
Manager also gets the user.collect message and invokes the UserCollected handler with the wrong data. (wrong)
Collector's CollectUser handler does work, then publishes a message with routingKey user.collected (correct)
Manager gets the user.collected message and invokes the UserCollected handler (correct)

我的问题

管理器为什么会得到user.collect消息，原因是:

My Question

Why does the Manager get the user.collect message, given:

它在COLLECTED_USER队列而不是COLLECT_USER队列上侦听，并且
正在COLLECT_USER队列上监听的 Collector 已经处理了该消息.

It's listening on the COLLECTED_USER queue not the COLLECT_USER queue, and
The Collector, which is listening on the COLLECT_USER queue, has already handled the message.

实施细节

我按如下方式创建订阅者和发布者(针对相关性进行了修剪)

Implementation details

I create the subscribers and publishers as follows (trimmed for relevance)

根据AMQP url和参数url，exchange，type，routingKey，queueName和handler

given the AMQP url and params url, exchange, type, routingKey, queueName and handler

const connection = await amqp.connect(url)
const channel = await connection.createChannel()
channel.assertExchange(exchange, type, { durable: true })
const result = await channel.assertQueue(queueName, { exclusive: false })
channel.bindQueue(result.queue, exchange, routingKey)
channel.prefetch(1)
channel.consume(result.queue, handler)

创建发布者

鉴于AMQP url和参数url，exchange和type

const connection = await amqp.connect(url)
const channel = await connection.createChannel()
await channel.assertExchange(exchange, type, { durable: true })

发布

给出channel和参数exchange，routingKey和message

await channel.publish(exchange, routingKey, message)

注意

此问题是.

Note

This question is a follow-on from RabbitMQ — Why are my Routing Keys being ignored when using topic exchange.

publishes