IBM MQ provider for JMS：如何自动回滚消息？

Question

应用程序中的工作版本

• IBM AllClient版本：'com.ibm.mq:com.ibm.mq.AllClient:9.1.1.0'
• org.springframework:spring-jms:4.3.9.发布
• javax.jms:javax.jms-api:2.0.1

我的要求是，如果由于消费者不可用（例如DB不可用）而导致消息处理失败，则消息保留在队列中或放回队列中（如果可能的话）。这是因为消息的顺序很重要，必须按照接收消息的相同顺序来消费消息。Java应用程序是单线程的。

@Override
public void onMessage(Message message)
{
   try{
      if(message instanceOf Textmessage)
      {
      }
   
      :

      : 
      throw new Exception("Test");// Just to test the retry
    }
    catch(Exception ex)
    {
            try
            {
                int temp = message.getIntProperty("JMSXDeliveryCount");
                throw new RuntimeException("Redlivery attempted ");
                // At this point, I am expecting JMS to put the message back into the queue.
                // But it is actually put into the Bakout queue.
            }
            catch(JMSException ef)
            {
                String temp = ef.getMessage();
            }

    }
}

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

如果将消息放回队列中不可行，那么如何浏览消息、处理它，如果成功，如何拉出消息（这样它就被消费了，不再在队列中）？IBM provider for JMS支持此场景吗？

IBM MQ本地队列具有bothresh(1)。

Answer 1

若要保留消息顺序，一种方法可能是作为回滚策略的一部分暂时停止消息侦听器。查看DefaultMessageListenerContainer的Spring Boot文档，有一个Stop(Runnable callback)方法。我已经尝试在回滚中使用它，如下所示。

为了确保我的侦听器是单线程的，我在DefaultJMSListenerContainerFactory上设置了ContainerFactory.setConcurrency（“1”）。

在我的侦听器中，我设置了ID

@JmsListener(destination = "DEV.QUEUE.2", containerFactory = "listenerTwoFactory", concurrency="1", id="listenerTwo")

JmsListenerEndpointRegistry reg = context.getBean(JmsListenerEndpointRegistry.class);
DefaultMessageListenerContainer mlc = (DefaultMessageListenerContainer) reg.getListenerContainer("listenerTwo");

retryCount = Integer.parseInt(msg.getStringProperty("JMSXDeliveryCount"));
if (retryCount < 5) {
    throw new Exception("Rollback test "+retryCount);
}

//catch processing here and decide to rollback
mlc.stop(new ContainerTimedRestart(mlc,delay));
System.out.println("#### "+getClass().getName()+" Unable to process message.");
throw new Exception();

public class ContainerTimedRestart implements Runnable {

  //Container instance to restart.
  private DefaultMessageListenerContainer theMlc;

  //Default delay before restart in mills.
  private long theDelay = 5000L;

  //Basic constructor for testing.
  public ContainerTimedRestart(DefaultMessageListenerContainer mlc, long delay) {
    theMlc = mlc;
    theDelay = delay;
  }

  public void run(){
    //Validate container instance.

    try {
      System.out.println("#### "+getClass().getName()+"Waiting for "+theDelay+" millis.");
      Thread.sleep(theDelay);
      System.out.println("#### "+getClass().getName()+"Restarting container.");
      theMlc.start();
      System.out.println("#### "+getClass().getName()+"Container started!");
    } catch (InterruptedException ie) {
      ie.printStackTrace();

      //Further checks and ensure container is in correct state.
      //Report errors.
    }
  }