java - ActiveMQ : dead letter queue keeps my messages order

Question

我使用 ActiveMQ 作为代理来传递消息。这些消息旨在写入数据库中。有时，数据库无法访问或已关闭。在这种情况下，我想回滚我的消息以便稍后重试此消息，并且我想继续阅读其他消息。

这段代码工作正常，除了一点:回滚消息阻止我阅读其他代码:

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;
}

我以这种方式创建我的 session :

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

回滚很容易问:

session.rollback();

假设我的队列中有 3 条消息:

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

我的消费者会做(线性序列):

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

但是我想要:

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)

那么，我该如何配置我的消费者以延迟我回滚的消息呢？

Answer 1

这实际上是预期的行为，因为消息重试是由客户端处理的，而不是代理。因此，由于您有 1 个 session 绑定(bind)，并且您的重试策略是为 DLQ 之前的 3 次重试设置的，因此整个重试过程会阻塞该特定线程。

所以，我的第一个问题是，如果数据库插入失败，您是否会期望所有其他数据库插入都因类似原因而失败？

如果不是，解决这个问题的方法是将该队列的重试策略设置为 0 次重试，并使用特定的 DLQ，这样消息将立即失败并进入 DLQ。然后让另一个进程每 5 秒退出 DLQ 并重新处理和/或将其放回主队列中进行处理。