如果在使用消息后出现任何错误,如何在JMS Message Queue中保留消息？

Question

我的方案是 - 我将消息发送到队列,一旦消息被消耗,我将它发送到第三方中间件应用程序.如果该中间件应用程序已关闭,那么我发布的消息就会消失.如果中间件应用程序关闭而我希望它处于保持状态或在队列中等待,我不想丢失该消息.请建议,如何处理这种情况？

Answer 1

您应该像这样创建会话:

Session session = connection.createSession(false,
                       Session.CLIENT_ACKNOWLEDGE);

当您尝试将邮件传递到第三方应用时:

• 如果它正在工作,你应该知道这个消息.

• 如果它失败了你就不应该知道它,这样JMS提供者就能够重新发送它,并且消息不会丢失. message.acknowledge();

另外,您可以看一下:JMS AUTO_ACKNOWLEDGE何时被确认？

Answer 2

JMS队列不是消息存储库.

如果您有一个"错误消息",处理继续失败,那么JMS服务器(如果已配置)将不可避免地将该消息转储到"死消息队列",这将慢慢填满,直到另一个进程耗尽它.

您不希望在队列中保留错误消息,因为它们可能会阻塞队列(假设您有10个消费者,并且前10个消息都是坏消息,因此所有进程都会继续查看错误消息 - 拖延队列).

因此,您需要一些机制将消息存储到异常接收器中,然后可以将它们注入主队列进行处理.

死消息队列不是这种机制(不将消息存储在JMS队列中),而是可以将ROUTE异常消息路由到更永久的存储区域(即db表或其他东西).

在那里,他们可以被审查(自动,手动,无论如何),并重新传递或取消.

但关键是你需要一个外部机制,单独的JMS服务器不适合这类消息.

Answer 3

这可以通过使用会话确认来实现。为此，首先修改生产者代码以使用 Session.AUTO_ACKNOWLEDGE。创建队列会话时，将 AUTO_ACKNOWLEDGE 设置为 false。这意味着消费者必须承认。当消费者发送消息确认时，该消息将从队列中删除，否则将保留在队列中。

下面是生产者代码。

try {
        QueueConnectionFactory qcf = AppUtils.getQueueConnectionFactory(); 
        Queue q = AppUtils.getDestination();
        QueueConnection qConnection = qcf.createQueueConnection();
        QueueSession qSession = qConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        QueueSender qSender = qSession.createSender(q);

        qConnection.start();
        TextMessage msg = qSession.createTextMessage("Hello");
        qSender.send(msg);

        qSender.close();
        qConnection.close();

    } catch (JMSException e) {
        // log your error to log file
        e.printStackTrace();
    }

在消费者方面，您必须执行相同的操作，创建一个 AUTO_ACKNOWLEDGE 为 false 的队列会话。

处理完消息后，您可以发送确认以从队列中删除该消息，否则该消息将保留在队列中。

try {
        QueueConnectionFactory qcf = getQueueConnectionFactory(); 
        Queue q = getDestination();
        QueueConnection qConnection = qcf.createQueueConnection();
        QueueSession qSession = qConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        QueueReceiver qReceiver = qSession.createReceiver(q);

        qConnection.start();
        Message msg = qReceiver.receive();

  // here send your message to third party application 
  //if your third party application is down 
        if(thirdpartyapp is down){
          //here you can raise an exception 
          //or just do nothing 
          // you're not sending acknowledgement here so the msg will 
           //remain in the queue
        }else{      
          msg.acknowledge();//here youre sending ack, so msg will be deleted
          qReceiver.close();
          qConnection.close();
        }

    } catch (JMSException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }