java - Apache Camel : Reply received for unknown correlationID

Question

其他两个软件组件之间有一个中间件。在中间件中，我通过Apache Camel路由Apache ActiveMQ消息。

它是这样工作的:

1. 1stComponent 使用 middleware 向 3rdComponent 发送消息

2. 3rdComponent 回复消息并将其发送回 1st(使用中间件)。

              1stComponent <<=>> Middleware <<=>> 3rdComponent
   

问题:

我在中间件中使用ConcurrentConsumers。

在顺序发送大量消息的过程中，突然中间件停止了所有进程!没有任何异常或消息!例如，500 条消息中的前 100 条已得到处理，其余消息作为待处理消息保留在队列中。

有时会在进程中间记录此警告:

[WARN ] TemporaryQueueReplyManager(Camel (camel-1) thread #11 - TemporaryQueueReplyManager[Q.MyQ]):91 - Reply received for unknown correlationID [c551c7aa061f501c]. The message will be ignored: ActiveMQMapMessage {commandId = 2161, responseRequired = true, messageId = ID:xxxxxxx, originalDestination = null, originalTransactionId = null, producerId = ID:xxxxxxx, destination = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, transactionId = null, expiration = 0, timestamp = 1389890272360, arrival = 0, brokerInTime = 1389890272360, brokerOutTime = 1389890272360, correlationId = c551c7aa061f501c, replyTo = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = org.apache.activemq.util.ByteSequence@19e19da, dataStructure = null, redeliveryCounter = 0, size = 0, properties = {breadcrumbId=ID:xxxxxxxxxxxxxx, Title=300, CamelJmsDeliveryMode=1}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false} ActiveMQMapMessage{ theTable = {} }

这是中间件代码:

private static class MyRouteBuilder extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("activemq:queue:Q.Middleware?concurrentConsumers=1&maxConcurrentConsumers=10")
        .threads(1, 100)
            .process(new Processor() {
                public void process(Exchange exchange) {
                    //some code
                }
            })
        .inOut("activemq2:queue:Q.3RD")
        ;
    }
}

和3rdComponent:

private static class MyRouteBuilder extends RouteBuilder {
    @Override
    public void configure() {
        from("activemq:queue:Q.3RD")
        .threads(1, 100)
        .process(new Processor() {
            public void process(Exchange exchange) {
                //some code
            }
        })
        ;
    }
}

Answer 1

@Amin Ralf 在他的回复中添加了更多内容，这是正确的 --- 进程停止的原因可能有两个1 - TTL 已过期2 - 客户端和服务器之间的时间不同步。

如果您的问题与 1 有关，则设置 header -“JMSExpiration”

如果您的问题与 2 -(从另一个 stackoverflow 帖子复制粘贴)相关

那么客户端和经纪人之间的时钟需要同步，以便到期正常工作。如果时钟不同步，则当代理收到消息时，客户端设置的到期时间可能已经到期。或者客户端时间领先于代理，因此到期时间比 10 秒长。

通过重新调整时间来解决此问题，仅基于经纪人。请参阅http://activemq.apache.org/timestampplugin.html