JMS的IBM MQ提供程序:如何自动回滚消息?

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

To preserve message ordering, one approach might be to stop the message listener temporarily as part of your rollback strategy. Looking at the Spring Boot doc for DefaultMessageListenerContainer there is a stop(Runnable callback) method. I've experimented with using this in a rollback as follows.

为了确保我的侦听器是单线程的，在我的DefaultJmsListenerContainerFactory上设置了containerFactory.setConcurrency("1").

To ensure my Listener is single threaded, on my DefaultJmsListenerContainerFactory I set containerFactory.setConcurrency("1").

在我的侦听器中，设置一个id

In my Listener, I set an id

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

并检索DefaultMessageListenerContainer实例.

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

为了进行测试，我检查了JMSXDeliveryCount并将异常抛出回滚.

For testing, I check JMSXDeliveryCount and throw an exception to rollback.

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

在侦听器的catch处理中，我在DefaultMessageListenerContainer实例上调用stop(Runnable callback)，并传入如下定义的新类ContainerTimedRestart.

In the Listener's catch processing, I call stop(Runnable callback) on the DefaultMessageListenerContainer instance and pass in a new class ContainerTimedRestart as defined below.

//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();

ContainerTimedRestart扩展了Runnable，并且DefaultMessageListenerContainer负责在stop调用完成时调用run()方法.

ContainerTimedRestart extends Runnable and DefaultMessageListenerContainer is responsible for invoking the run() method when the stop call completes.

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

我在队列中加载了三个带有有效负载"a"，"b"和"c"的消息.分别启动监听器.

I loaded my queue with three messages with payloads "a", "b", and "c" respectively and started the listener.

在我的队列管理器上检查DEV.QUEUE.2，我看到IPPROCS(1)确认只有一个应用程序句柄打开了队列.每次滚动五次后，将按顺序处理这些消息，并且两次回滚尝试之间的延迟为5秒.

Checking DEV.QUEUE.2 on my queue manager I see IPPROCS(1) confirming only one application handle has the queue open. The messages are processed in order after each is rolled five times and with a 5 second delay between rollback attempts.