java - 根据消息属性暂时禁用 JMS 消息的传递

Question

我有一个要求，目前我不知道是否可能。如果消息包含指定的属性，我想暂时禁用 JMS 消息的开发。目前我使用 HornetQ 作为消息提供程序。

让我们举个例子:

队列包含以下三个条目:

{1, "foo", "A_CATEGORY"}
{2, "bar", "B_CATEGORY"}
{9, "bof", "A_CATEGORY"}

在某个时刻，应用程序必须能够告诉 HornetQ 消息服务器此时不应传递属于 B_CATEGORY 的消息(例如，因为 B_CATEGORY 对象的基础数据库已更新)。因此，ID 为 2 的消息目前不会被传递，而 1 和 9 将会被传递，因为它们对于类别对象具有不同的值。

它必须在 Java 代码之外发生，根本不需要重新启动应用程序。这可能吗？

感谢您的帮助!

<小时/>

只是考虑了解决此问题的替代设计方法。我们假设第一个队列包含各种类别的消息(顺便说一句，不可能为每个类别创建一个队列，因为可能有很多类别)。这个“正常”队列是正常配置的(例如，没有过期时间，但有 DLQ)。

现在，如果监听器使用此类消息并发现它无法处理属于特定类别的消息，则会将其放入第二个队列中。该队列配置了重新传递延迟和到期时间。如果现在将到期时间设置得足够高(当然不是队列溢出)并且重新传递时间不太短，那么如果上述问题没有解决方案，这应该可以解决。

当然，必须计算在无法处理某个类别的时间内可以创建多少个队列条目。以及某个类别的不可用可能需要多长时间，以便可以相应地调整重新交付。

Answer 1

据我所知，消息驱动的 Bean 是不可能的。

使用标准 JMS 使用者可以实现类似的功能:

 MessageConsumer c = session.createConsumer(destination);
 while ( b-category-can-be-processed ) {
     Message m = c.receive();
     // process messages until b category is OK to be processed
 }

 c.close();

 // now create a different consumer with message selector ignoring "B_CATEGORY"
 MessageConsumer c1 = session.createConsumer(destination, "Category <> 'B_CATEGORY'");
 while ( b-is-locked ) {
     Message m = c1.receive();
     // process messages until b category is locked
 }

 c1.close();
 // go to start

此示例假设您能够根据收到的消息判断何时再次处理 B。如果没有，那么您可以在一段时间后恢复正常的日常生活。该示例还仅呈现单个执行线程。

进一步探索这条路径，您可以看看 Spring 的 DefaultMessageListenerContainer — Spring 消息驱动 bean。它可以完全按照我所描述的方式进行操作，但方式要高级得多。它可以通过消息选择器提供，并且它是实时的，您可以随时更改它。如果您将 concurrentConsumers 设置为大于 1，它也会处理多个线程中的消息。

至于您在无法处理消息时将消息重定向到另一个队列的解决方案，请注意它会产生额外的流量；您确实希望最终处理所有消息，对吧？为什么不把它们留在原处并在适当的时候取回它们呢？您不必提前估计重新投递延迟，这可能很困难。