如果使用延迟，Apache camel 不会将所有消息转发到 AMQ

Question

我正在尝试使用 apache camel 从 activemq 队列“AMQ:ORIGIN”中读取消息。读取消息后需要将其传递给两个不同的“AMQ 队列”。但条件如下。

1. 消息应立即传递到队列“AMQ:A”。
2. 消息应在延迟一分钟后传递到队列“AMQ:B”。

为了实现上述目标，我创建了两条路线。在第一条路线中，我从 AMQ 队列中读取数据，并对“AMQ:A”和“seda:delay”队列进行多播。在第二条路线中，我正在读取“seda:delay”队列，延迟一分钟，然后传递到“AMQ:B”队列。

如果将 1 或 10 条消息传递给“AMQ:ORIGIN”，则工作正常

如果我同时向“AMQ:ORIGIN”队列发送 100 条消息，那么

1. 所有 100 条消息都传递到“AMQ:A”队列
2. 只有 10 或 12 条消息被传递到“AMQ:B”队列。休息只停留在路线上。

以下是我的路线。

        <route id="read-origin">
            <from uri="activemq:ORIGIN"/>
            <multicast stopOnException="true">
                <to uri="activemq:A"/>
                <to uri="seda:delay-route"/>
            </multicast>
        </route>

        <route id="delay-route">
            <from uri="seda:delay-route"/>
            <delay asyncDelayed="true">
                <constant>60000</constant>
            </delay>
            <to uri="activemq:B"/>
        </route>

请提出上述更改。

谢谢，

Answer 1

这似乎很明显，因为您将每条消息延迟 1 分钟。

如果您将 100 条消息发送到 ORIGIN 队列，则需要 100 分钟，直到所有这些消息到达队列 @9​​87654323@。

第一条消息立即被消费，并延迟 1 分钟。第二个是在第一个交付时采取的（假设 seda 队列中有 1 个消费者）并且延迟一分钟等等......

我假设您希望已经在队列中等待 1 分钟的消息在使用时可以立即发送。

您可以轻松实现这一点，从而使延迟动态化。

实现一个计算消息的JMSTimestamp 标头（入队时间）与当前时间之间的差异的bean。

currentTime - JMSTimestamp = 已经等待

Your minimal delay - alreadyWaited = 投递前的等待时间（对于排队时间超过延迟的消息的负值取 0）

将此差异用作延迟值（我使用 Java DSL，因为我更了解它）。

from("seda:delay-route").routeId("delay-route")
    .delay().expression(method(YourDelayCalculationBean.class))
    .to("activemq:B");

像这样，如果你的消息堆积在队列中，它们可能都在消费时立即交付，因为它们已经在队列中等待超过 1 分钟。

因评论而添加

好的，抱歉，我没有发现 asyncDelayed。

文档对asyncDelayed 的评价听起来如您所愿。但根据您的评论，听起来Delay EIP 不再阻止消费者，而是阻止自己。

所以 seda 消费者接收一条消息，将其交给延迟并继续处理下一条消息。在 10 条消息之后（10 个线程是 Camel 的默认线程池大小），Delay 为“满”（所有线程都在固定的 1 分钟延迟内阻塞）。

因此，消费者被阻塞，因为延迟无法接收更多消息。一分钟后，延迟可以传递第一条消息并继续。

这只是根据您编写的路线行为方式进行的疯狂猜测。