使用多线程侦听器处理 JMS 事务和重新交付

Question

我正在使用JMS在Java 1.8 SE环境中处理消息。 这些消息来自 Oracle 高级队列。 由于处理消息可能需要一段时间，因此我决定拥有一个包含 5 个工作线程(MessageHandler 对象)的池，以便多个线程可以同时处理消息。 我希望保证传递，没有重复的消息传递。

我使用

queueConnection.createQueueSession(true, Session.SESSION_TRANSACTED);

以创建队列会话。 我使用以下代码来处理传入的消息。 基本上，onMessage生成一个处理消息的线程。

public class JmsQueueListener implements MessageListener
{
/** A pool of worker threads for handling requests. */
private final ExecutorService pool;
OracleJmsQueue queue;
public void onMessage(Message msg)
{
pool.execute(new MessageHandler(msg));
// can't commit here - the thread may still be processing
}
/**
* This class provides a "worker thread" for processing a message
* from the queue.
*/
private class MessageHandler implements Runnable {
/**
* The message to process
*/
Message message;
/**
* The constructor stores the passed in message as a field
*/
MessageHandler(Message message) {
this.message = message;
}
/**
* Processes the message provided to the constructor by
* calling the appropriate business logic.
*/
public void run() {
QueueSession queueSession = queue.getQueueSession();
try {
String result = requestManager.processMessage(message);
if (result != null) {
queueSession.commit();
}
else {
queueSession.rollback();
}
}
catch (Exception ex) {
try {
queueSession.rollback();
}
catch (JMSException e) {
}
}
}
}   //  class MessageHandler

我的问题是我不知道如何向原始队列指示处理是否已成功完成。我无法在onMessage结束时提交，因为线程可能尚未完成处理。 我不认为我目前有commits和rollbacks的地方也没有任何好处。 例如，如果 5 个工作线程处于不同的完成状态，则正在提交的队列会话的状态是什么？

我想我一定错过了一些关于如何以多线程方式处理 JMS 的基本概念。 任何帮助将不胜感激。

Answer 1

您使用的是异步消息处理，因此，除非您实现一种确保每个消息处理按时间顺序完成的方法，否则您最终将出现较旧的消息处理在最近的消息处理之后完成的情况。那么为什么要使用消息传递服务呢？

问题的简单解决方案是在onMessage方法的末尾commit，并在messageHandler正文中重新排队，以防出现错误。但是，如果重新排队本身失败，则此解决方案可能会出现问题。