ActiveMQ 中的 AMQP 1.0 支持 - 无法设置预取答案

【问题标题】：AMQP 1.0 support in ActiveMQ - can't set prefetchActiveMQ 中的 AMQP 1.0 支持 - 无法设置预取
【发布时间】：2014-07-30 22:18:01
【问题描述】：

我们正在尝试在 ActiveMQ 5.9.1 之上构建一个作业队列系统。我们在 activemq 中使用 AMQP 1.0 支持，目前使用 qpid jms 库作为客户端代码。

对于作业队列系统，我们可以将 prefetch 设置为 1 至关重要 - 我们希望我们的作业工作者一次只能获得一个任务，竞争消费者模式。 qpid 库使用连接工厂定义来支持这一点：

connectionfactory.activemq-amqp-manager = amqp://{user}:{password}@{hostname}:{port}?clientid=job-manager&remote-host=default&max-prefetch=1

但是，ActiveMQ 似乎完全忽略了这一点，消费者在 activemq 管理控制台中显示为具有 100 的预取。

进一步挖掘，我找到了来源：org.apache.activemq.transport.amqp.AmqpProtocolConverter

其中包括以下行：

  consumerInfo.setPrefetchSize(100);

这是否意味着 AMQP 1.0 协议的 ActiveMQ 支持硬连线了所有客户端的预取值 100？显然这是非常错误的。我们有哪些选择？我们可以切换到纯 JMS 客户端，但会失去语言互操作性。或者我们可以切换到 RabbitMQ，这意味着引入另一个产品来支持（ActiveMQ 已经在其他地方使用）。

【问题讨论】：

标签： activemq amqp prefetch qpid

【解决方案1】：

您是否实际进行过任何测试，以查看您的预取为 1 的消费者是否按预期工作？

ActiveMQ 的内部结构并非基于 AMQP，因此即使 AMQP 使用者已连接，事情也不会总是如您所愿。 AMQP 客户端连接的 ActiveMQ 内部消费者默认为 100，但是向客户端发送消息仍然受到远程端设置的当前流窗口的限制。如果 QPid 客户端在您设置预取值时将链接信用限制为 1，那么代理将只会向其发送一条消息。

这里唯一需要注意的是，每个订阅最初都可以在代理端预取，但最终会在消费者之间平均分配，因为随着更多消费者的添加，代理将进行循环调度。

ActiveMQ 中的 AMQP 支持仍在成熟中，因此欢迎您潜入并尝试改进它。

【讨论】：

嗨，是的，我们已经测试过了。例如 - 将 5 条消息放在一个队列中。启动一个工人/消费者 - 它获取第一条消息并开始处理它。启动第二个工人/消费者，它坐在那里什么都不做。同时，第一个工作人员在处理完第一条消息后，将获得第二条和第三条等。看起来很明显，第一个工作人员的客户端已经有效地预取了队列中的所有消息，尽管 max-prefetch 设置在 qpid 中，结合 activeMQ 管理上下文显示消费者具有 100 的预取。
是的，这是一个已知的限制。我在 AMQP 支持方面做了一些工作，但还没有时间解决这个问题。欢迎投稿。
感谢您的确认。将看看它是多么容易修补。