发布消息前选择消费者rabbitmq

Question

我正在尝试构建一个系统，我需要选择下一个可用且合适的消费者从队列发送消息（或者可能是任何其他不使用队列的解决方案）

要求 我们有多个发布者/客户将发送对象（图像）以在一侧进行处理，并且有多个分析师将处理它们，一旦处理，发布者应该得到相应的响应。

发布者并不关心哪个分析师将处理数据。

用户有一个网络应用程序，他们可以将每个客户端/发布者映射到一个或多个或所有代理，例如，如果发布者 P1 映射到代理 A 和 B，则来自 P1 的所有对象都可以由代理 A 或Agent B. 注意：一个对象只能由一个代理处理。

根据映射，我应该有一个中间件，它使用来自所有发布者的消息并分发给代理

解决方案 1 我最初的想法是建立一个所有发布者发布消息的队列。代理发布消息说他们正在等待处理对象的另一个队列。

中间件选择消息，获取可以将消息发送到的可能代理列表（从缓存的数据库中），并通过代理队列找到下一个合适且可用的代理并将消息发布到该代理。

这个解决方案的问题是，如果我有像 a、b、c、d 这样的代理队列，而我收到的消息只能由代理 b 处理，我将拒绝代理 d & c，他们最终会排在最后我有大约 180 个代理，所以他们可能永远不会被选中，或者如果下一条消息只能由代理 d 处理（例如），我们必须拒绝所有代理才能到达那里

解决方案 2 从发布者到中间件的第一位还是一样

拥有一个可扩展的快速 nosql 数据库，代理会在其中添加一条记录以通知那里的可用性。基本上是一个键值对

中间件从缓存中获取配置，并从 nosql 数据库中获取下一个可用 + 合适的代理 将消息发送到代理的队列（通过直接交换）并更新 nosql 以设置 isavailable false ad 获取下一条消息。

这个解决方案的问题是数据库和中间件可能成为瓶颈，如果我扩展中间件，我最终会遇到数据库并发问题，例如我有两个中间件副本正在运行，每个副本都会收到一条可以处理的消息由代理 A 和 B 提供，两个代理均可用。 这两个中间件副本将查询数据库，并可能使 A 可用，并最终将两条消息发送给 A，而 B 仍在等待消息处理。

我将拥有大约 100 个出版商和 180 个代理商。

任何想法如何改进这些解决方案或任何其他可行的解决方案将不胜感激？

基于此，我还需要弄清楚代理如何将响应发送回发布者。

谢谢

Answer 1

我将从我的开源服务总线的角度来回答这个问题：Shuttle.Esb

通常会忽略任何基于内容的路由，而只是使用分发者模式。所有消息都转到主端点，它将分发消息。但是，如果您决定坚持使用这些逻辑分组，则可以为每个逻辑分组（每个代理组）设置主要端点。您仍将拥有主端点，但不是将工作人员端点映射到代理，而是将代理分组映射到逻辑主端点，工作人员支持 that。

然后在主端点中，您将根据您的内容（作为代理标识符）将消息转发到相关的逻辑主端点。您一直在跟踪原始发件人。然后，在工作人员中，您会将消息发送回原始发送者的队列。

我相信您可以使用任何服务总线进行几乎相同的操作。

Answer 2

我在这里看到了几个要求，我认为可以归结为几件事：

• 发布者不关心哪个代理处理图像
• 发布者需要知道图像处理何时完成
• 代理一次只能处理一张图片
• 代理只能处理某些图像

这些假设是否正确？我错过了什么重要的事情吗？

如果不是，那么您的解决方案几乎内置于带有路由和队列的 RabbitMQ。应该不需要构建自定义中间层服务来管理它。

使用 RabbitMQ，您可以将使用者设置为一次仅处理 1 条消息。消费者将其“预取”限制设置为 1，并从队列中检索一条“无确认”设置为 false 的消息——这意味着，它必须在处理完消息后确认该消息。

要仅使用特定代理可以处理的消息，请将 RabbitMQ 的路由功能与多个队列一起使用。队列将根据图像类型或消费者可以选择图像的其他标准创建。

例如，如果有两种类型的图像：TypeA 和 TypeB，您将有 2 个队列 - 一个用于 TypeA，一个用于 TypeB。

那么，如果 Agent1 只能处理 TypeA 图像，它只会从 TypeA 队列中消费。如果 Agent2 可以处理这两种类型的图像，它会从两个队列中消耗。

要将正确的图像放入正确的队列中，发布者需要使用正确的路由键。如果您知道图像类型（或任何选择标准），您将更改发布者端的路由键以匹配该选择标准。 RabbitMQ 中的路由将设置为将 TypeA 的消息移动到 TypeA 队列等。

最后一部分是在图像处理完成时得到响应。这可以通过 RabbitMQ 的“回复”字段和相关代码来完成。它的要点是发布者有自己的排他队列。当它发布一条消息时，它会在消息的“回复”标头中包含它的独占队列的名称。当代理完成图像处理后，它会通过“回复”标头中的队列发回状态更新消息。该状态更新消息告诉生产者请求的状态。

从 RabbitMQ 的角度来看，可以使用此处的示例和文档将这些部分组合在一起：

http://www.rabbitmq.com/getstarted.html

具体看这些：

• 工作队列：http://www.rabbitmq.com/tutorials/tutorial-two-python.html
• 主题：http://www.rabbitmq.com/tutorials/tutorial-five-python.html
• RPC（又名请求/响应）：http://www.rabbitmq.com/tutorials/tutorial-six-python.html

您可以在这些文档中找到多种语言的示例。

我还在my RabbitMQ Patterns eBook 中介绍了大多数这些场景（和其他场景）

Answer 3

由于发送者和接收者的总数只有数百个，如何为每个发送者创建一个队列。根据您的发送者接收者映射，接收者订阅发送者队列（更新订阅映射更改）。您可以将接收器配置为仅在完成处理一条消息时（以随机方式）从它订阅的所有队列中接收下一条消息。