Apache Camel: resposta recebida por correlação desconhecida
Existe ummiddleware
entre dois outros componentes de software. Nomiddleware
Estou encaminhandoApache ActiveMQ
mensagens deApache Camel
.
é assim que funciona:
1stComponent
usamiddleware
para enviar mensagem para o3rdComponent
3rdComponent
responde a mensagem e a envia de volta para o1st
(usandomiddleware
)
1stComponent <<=>> Middleware <<=>> 3rdComponent
Problema:
estou a usarConcurrentConsumers
no middleware.
No meio do envio de muitas mensagens sequencialmente, de repentemiddleware
pára todo o processo! não há exceções ou mensagens! por exemplo, as primeiras 100 das 500 mensagens foram processadas e o restante permanece na fila como mensagens pendentes.
esse aviso é registrado algumas vezes no meio do processo:
[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 = {} }
Isto é omiddlewares
Código:
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")
;
}
}
e a3rdComponent
:
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
}
})
;
}
}