java - 使用 Apache Camel 处理不同优先级的 JMS 消息

Question

我正在解决处理不同优先级的JMS消息的问题。我使用 Apache Camel 框架作为 EIP 实现者。我有两个必须从中使用的队列。第一个包含具有较高优先级的消息，第二个包含具有较低优先级的消息。

现在我想创建一个特殊的组件“混合器”，其中包含执行以下步骤的循环:

1. 查看优先级队列
   • 如果包含消息，则将其发送到输出队列并进入循环的开始(如果包含大量消息，则将处理前10条消息，然后进入步骤2)
   • 如果不包含任何消息，请转到步骤 2
2. 查看第二个队列
   • 如果它包含一条消息，则将其发送到输出队列(但只有一条消息)并重复循环
   • 如果不包含任何消息，则重复循环

正如你所看到的，如果有很多消息，我想保持一个特殊的比率(优先级队列中的 10 条/第二条消息中的 1 条)。如果优先队列中没有消息，我们可以立即处理第二队列中的消息。我想要像 EIP Resequencer 这样的东西，它可以从多个输入队列中消费。

我寻找了一个 Camel 的路线，我将从两个队列中消费，其中我将有一个上面描述的组件“混合器”。我想要这样的东西:

<route>
  <from id="A" />
  <from id="B" />
  <resequence id="mixer" />
  <to id="C" />
</route>

但是我没有找到方法。一条路由只能有一个输入。如果有更多输入，Camel 将在内部复制路线(我们将有两个独立的路线)，并且行为将如下:

<route>
  <from id="A" />
  <resequence id="mixer" />
  <to id="C" />
</route>

<route>
  <from id="B" />
  <resequence id="mixer" />
  <to id="C" />
</route>

as descibed here 。这不是我想要的:-(。你有什么想法如何解决我的问题吗？提前致谢!

Answer 1

也许你可以看看 Polling Consumer生态工业园区。

我展示了来自camel网站的一些示例代码，这可能是您解决方案的基础:

public void someBusinessLogic() {
    // loop to empty queue
    while (true) {
        // receive the message from the queue, wait at most 3 sec
        String msg = consumer.receiveBody("activemq:queue.inbox", 3000, String.class);
        if (msg == null) {
            // no more messages in queue
            break;
        }

        // do something with body
        msg = "Hello " + msg;

        // send it to the next queue
        producer.sendBodyAndHeader("activemq:queue.foo", msg, "number", count++);
    }
}

您可以尝试修改此代码，让计数器尝试消耗 hi-pri 队列中的 10 条消息和 lo-pri 队列中的 1 条消息。

玩得开心，祝你好运。有兴趣知道这是否成功:)