2

为了澄清标题：我使用的是 ActiveMQ 5.15.15（不是 Artemis 引擎），并且我使用的是没有官方 JMS 库的 AMQP 1.0。更具体地说，我使用的是 AmazonMQ 版本，它将很快升级到 5.16.2。如果需要，我可以强制升级。

我正在使用一个AMQP 1.0 兼容库(rhea)，到目前为止它为我们提供了很好的服务，但我没有找到任何关于如何让 ActiveMQ 的重新交付插件与我的库一起使用的文档。库维护者也不知道它是如何通过 ActiveMQ 公开的。

尽管尝试添加各种标头、传递注释、消息注释或应用程序属性，但我无法让重新传递插件工作。我schedulerSupport="true"的代理元素中有服务器配置。

这些是我尝试过的键，值是数字。例如，30000在允许消费者/订阅者看到队列中的消息之前的 30 秒。我在各种文档中看到了它们，尝试它们并没有什么坏处。

  • AMQ_SCHEDULED_DELAY
  • x-opt-delivery-delay
  • _AMQ_SCHED_DELIVERY

我还release收到了来自客户端的消息，这意味着它无法传递（还将一个值传递给代理失败并增加尝试传递计数）。虽然交付尝试的数量增加了，但延迟和指数退避似乎并没有在经纪人层面发挥作用。

我看到 STOMP 协议在发布时允许使用标头，这允许更清楚地设置选项。但是，除非这样做有意义，否则我不想切换所有内容。

我还看到了另一种通过 REST API 将延迟消息作为主题发送的功能，但我不确定这是否旨在成为生产用例。

所以现在，我要么在看：

  1. 将消息在内存中保留一段时间，并在延迟后尝试重新发布或释放它
  2. 调查STOMP，看看redelivery插件是否适用于它

但我希望有人知道在哪里实现这一点。

我的 redeliveryPolicy 是基本的：


               <!--
               The Redelivery plugin extends the capabilities of destination policies with respect to message redelivery.
               For more information, see http://activemq.apache.org/message-redelivery-and-dlq-handling.html
               -->
               <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
                 <redeliveryPolicyMap>
                   <redeliveryPolicyMap>
                     <redeliveryPolicyEntries>
                       <!--<redeliveryPolicy maximumRedeliveries="4" queue="SpecialQueue" redeliveryDelay="10000"/>-->
                     </redeliveryPolicyEntries>
                     <defaultEntry>
                       <!-- 5s -> 15s -> 45s -> 135s -> 405s -->
                       <redeliveryPolicy backOffMultiplier="3" initialRedeliveryDelay="5000" maximumRedeliveries="5"/>
                     </defaultEntry>
                   </redeliveryPolicyMap>
                 </redeliveryPolicyMap>
               </redeliveryPlugin>

更新

我正在使用 auth 插件，并且有一个条目似乎是用于内置进程的。我认为这来自示例/默认配置。快速搜索似乎没有很多关于此的文档。我可以尝试向其他用户开放访问权限，但在当前设置下，每次更新/重新启动最多可能需要 15 分钟。

<authorizationEntry admin="administrators" queue="SchedulingProcessor.>" write="scheduling-processor"/>

评论澄清

  • 我的主要目标是延迟重新传递，因此消费者不会看到被放回队列 n 秒内的失败消息。
  • 我从没有特殊的标题/属性/注释+重新交付插件开始，这也不起作用。
4

1 回答 1

0

消息传递延迟和消息重新传递延迟之间存在区别，我认为您对此感到困惑，或者至少问题混淆了。

发件人可以使用已发送消息的“消息注释”部分并添加“x-opt-delivery-delay”或“x-opt-delivery-time”注释，从 AMQP 客户端请求在一些延迟后传送消息，假设经纪人已启用预定交货。这方面的一些例子可以在 ActiveMQ 单元测试中找到。延迟注释表示从接收时间开始的相对延迟（以毫秒为单位），而传递时间注释表示以 UTC 格式传递消息的时间。

ActiveMQ 5 重新传递策略会影响已在客户端上明确标记为不可传递的消息，因此 AMQP Released 结果不是触发此行为的正确选择，因为它只是表明客户端不会处理它并且远程应将其视为未送达并将其发送到其他地方。您需要使用Rejected或Modified(undeliverableHere=true)之一来“毒化”" 消息并触发重新传递策略。如果事情进展顺利，这应该会在一些延迟后触发重新传递，尽管由于 ActiveMQ 5 有一个相对基本的 AMQP 协议头，即使您在此处明确设置了无法传递，它也可能会重新发送给同一个消费者旗子。我不知道该位已经测试了多少（如果有的话），因此您的里程可能会有所不同。

于 2021-09-09T17:43:16.450 回答