限制将持久消息批量发送到队列的 ActiveMQ 创建者的正确方法是什么？

Question

我有一个生产者，它利用JMS事务将持久消息批量发送到队列。

我已经进行了测试，发现当使用批处理大小为1时，会应用Producer Flow Control。我可以看到我的生产者正在按照我为队列配置的内存限制进行限制。这是我的Producer Flow Control配置：

<policyEntry queue="foo" optimizedDispatch="true"
     producerFlowControl="true" memoryLimit="1mb">
</policyEntry>

队列中待处理消息的数量是可控的，我认为这是生产者流控制运行的证据。

但是，当批处理大小增加到2时，我发现此内存限制不受尊重，并且生产者根本没有受到限制。证据是队列中挂起的消息数继续增加，直到达到配置的 storeUsage 限制。

我知道这可能是因为当批处理大小大于1时，消息以异步方式发送，即使我没有显式地将useAsyncSend设置为true。

ActiveMQ的生产者流控制文档提到，为了抑制异步发布者，我们需要在生产者中配置生产者窗口大小，一旦达到窗口限制，这将强制生产者等待确认。

但是，当我在生产者中配置生产者窗口大小并尝试批量发送消息时，会引发异常并且没有发送消息。

这让我思考并提出了一个问题，“在批量发送持久消息时，是否可以配置生产者窗口大小？”。

如果不是，那么什么是正确的方法来限制批量发送持久消息的生产者？

Answer 1

我在v5.8.0中发现了这个问题，但发现这个问题在v5.9.0及更高版本中得到了解决。从 v5.9.0 开始，我发现 PFC 开箱即用，即使对于异步发送消息的生产者也是如此。自批量发送（其中批量大小

我的制作者在没有任何额外配置的情况下被限制了。队列中的挂起邮件数没有增加，并且处于控制之下。

通过一个简单的Camel消费者消费消息（并将它们附加到文件中），我发现在v5.8.0中（我遇到了这个问题），我可以在36秒内发送有效负载2k的100k消息。但是它们中的大多数最终都是待处理的消息。

但在v5.9.0中，发送同一组消息以证明PFC所起的作用需要176秒。在我的例子中，挂起的消息数量从未超过1000。

我还使用v5.10.0和v5.12.0（撰写本文时的最新版本）进行了测试，它们按预期工作。

所以如果你面临这个问题，很可能你正在运行ActiveMQ v5.8.0或更早版本。简单地升级到最新版本应该可以解决这个问题。

我感谢非常有帮助的ActiveMQ邮件列表人员的建议和帮助。

谢谢你的回答。对不起，我没有在我的问题中提到我使用的版本，否则我相信您可以立即发现问题。

Answer 2

没有真正的方法来限制“每秒最大消息数”或类似的东西。您要做的是启用生产者流控制和vm游标，然后将该队列(如果您愿意，也可以是所有队列)的内存限制设置为某个合理的级别。

您可以在配置中决定当达到队列内存限制时，生成器是应该挂起还是抛出异常。

<policyEntry queue="MY.BATCH.QUEUE" memoryLimit="100mb" producerFlowControl="true">
  <pendingQueuePolicy>
    <vmQueueCursor/>
  </pendingQueuePolicy>
</policyEntry>