ActiveMQ: la cola de mensajes no entregados guarda la orden de mis mensajes

Question

Utilizo ActiveMQ como intermediario para entregar mensajes. Estos mensajes están destinados a ser escritos en dabatase. A veces, la base de datos es inalcanzable o inaccesible. En ese caso, quiero deshacer mi mensaje para volver a intentar más tarde este mensaje y quiero continuar leyendo otros mensajes.

Este código funciona bien, excepto en un punto: el mensaje rollback me está bloqueando la lectura de los otros:

private Connection getConnection() throws JMSException { 
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy(); 
    redeliveryPolicy.setMaximumRedeliveries(3); // will retry 3 times to dequeue rollbacked messages 
    redeliveryPolicy.setInitialRedeliveryDelay(5 *1000); // will wait 5s to read that message 

    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url); 
    Connection connection = connectionFactory.createConnection(); 
    ((ActiveMQConnection)connection).setUseAsyncSend(true); 
    ((ActiveMQConnection)connection).setDispatchAsync(true); 
    ((ActiveMQConnection)connection).setRedeliveryPolicy(redeliveryPolicy); 
    ((ActiveMQConnection)connection).setStatsEnabled(true); 
    connection.setClientID("myClientID"); 
    return connection; 
} 

Creo mi sesión de esta manera:

session = connection.createSession(true, Session.SESSION_TRANSACTED); 

Rollback es fácil pedir :

session.rollback(); 

Imaginemos tengo 3 mensajes en mi cola:

1: ok 
2: KO (will need to be treated again : the message I want to rollback) 
3: ok 
4: ok 

Mi consumidor hará (secuencia lineal):

commit 1 
rollback 2 
wait 5s 
rollback 2 
wait 5s 
rollback 2 
put 2 in dead letter queue (ActiveMQ.DLQ) 
commit 3 
commit 4 

Pero quiero:

commit 1 
rollback 2 
commit 3 
commit 4 
wait 5s 
rollback 2 
wait 5s 
rollback 2 
wait 5s 
put 2 in dead letter queue (ActiveMQ.DLQ) 

Entonces, ¿cómo se puede configurar el Consumidor para retrasar mis mensajes rollback más adelante?

Answer 1

Esto es en realidad el comportamiento esperado, debido a reintentos de mensajes son manejados por el cliente, no el corredor. Entonces, dado que tiene 1 sesión enlazada, y su política de reintento está configurada para los 3 intentos antes de DLQ, entonces todo el proceso de reintento bloquea ese hilo en particular.

Entonces, mi primera pregunta es que si la inserción de la base de datos falla, ¿no esperaría que todas las inserciones de la base de datos fallaran por un motivo similar?

Si no, la forma de evitar esto es establecer 0 la política de reintento para esa cola de 0 reintentos, con un DLQ específico, para que los mensajes fallen inmediatamente y entren en el DLQ. Luego, tenga otro proceso que se retire del DLQ cada 5 segundos y reproceses y/o vuelva a colocarlo en la cola principal para su procesamiento.

Answer 2

¿Está utilizando el <strictOrderDispatchPolicy /> en el archivo de configuración XML de ActiveMQ? No estoy seguro de si esto afectará el orden de los mensajes para su reenvío o no. Si está utilizando un despacho de pedido estricto, intente comentar esa política para ver si eso cambia el comportamiento.

Bruce

Answer 3

Tuve el mismo problema, no he encontrado una solución aquí, así que decidí publicarlo aquí después de encontrar uno para las personas que luchan con lo mismo. Este se fija antes de la versión 5.6 cuando se establece nonBlockingRedelivery propiedad en true en la fábrica de conexiones:

<property name="nonBlockingRedelivery" value="true" />