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

Question

应用中的工作版本

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

我的要求是，如果由于消费者不可用（例如数据库不可用）而导致消息处理失败，则消息保留在队列中或放回队列中（如果这甚至可能的话)。这是因为消息的顺序很重要，消息必须按照接收到的顺序被消费。 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();
            }

    }
}

我已经在我的 spring.xml 中为 jmsContainer bean 设置了这个。

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

上面的代码有什么问题？

如果将消息放回队列中是不切实际的，如何浏览消息、处理它，如果成功，拉取消息（因此它被消费并且不再在队列中）？ IBM Provider for JMS 是否支持这种场景？

IBM MQ 本地队列有 BOTHRESH(1)。

Answer 1

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

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

在我的 Listener 中，我设置了 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 并抛出异常以回滚。

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

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

//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 负责在停止调用完成时调用run() 方法。

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”，然后启动了监听器。

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

Answer 2

用于 JMS 的 IBM MQ 类具有内置的有害消息处理。此处理基于 QLOCAL 设置 BOTHRESH，它代表 Backout Threshold。每个 IBM MQ 消息都有一个称为 MQMD（MQ 消息描述符）的“标头”。 MQMD 中的字段之一是BackoutCount。新消息的 BackoutCount 的默认值为 0。每次消息回滚到队列时，此计数都会增加 1。回滚可以来自对 rollback() 的特定调用，或者由于应用程序在调用 commit() 之前与 MQ 断开连接（由于网络例如问题或应用程序崩溃）。

如果您设置BOTHRESH(0)，则会禁用毒药消息处理。

如果BOTHRESH >= 1，则启用有害消息处理，当用于 JMS 的 IBM MQ 类从队列中读取消息时，它将检查 BackoutCount 是否 >= 到 BOTHRESH。如果消息符合有害消息处理的条件，那么它将被移动到BOQNAME 属性中指定的队列中，如果此属性为空或应用程序由于某种原因无权访问此队列，它将改为尝试将消息放入队列管理器DEADQ 属性中指定的队列，如果无法放入其中任何一个位置，则将回滚到队列中。

---

您可以在 IBM MQ v9.1 知识中心页面Developing applications>Developing JMS and Java applications>Using IBM MQ classes for JMS>Writing IBM MQ classes for JMS applications>Handling poison messages in IBM MQ classes for JMS中找到有关用于 JMS 毒消息处理的 IBM MQ 类的更多详细信息

Answer 3

在 Spring JMS 中，您可以定义自己的容器。为一个 Jms Destination 创建一个容器。我们应该运行一个单线程的 JMS 监听器来维护消息的顺序，让这项工作将并发设置为 1。

我们可以将容器设计为一旦遇到错误就返回 null，失败后所有接收调用都应该返回 null，这样在目标再次处于活动状态之前，不会从目标轮询任何消息。我们可以使用时间戳来保持活动状态，这可能是简单的毫秒。一个示例 JMS 配置应该足以添加退避。您可以添加小睡眠，而不是从receiveMessage方法连续返回null，例如在进行下一次调用之前睡眠10秒，这样可以节省一些CPU资源。

@Configuration
@EnableJms
public class JmsConfig {

  @Bean
  public JmsListenerContainerFactory<?> jmsContainerFactory(ConnectionFactory connectionFactory,
      DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory() {
      @Override
      protected DefaultMessageListenerContainer createContainerInstance() {
        return new DefaultMessageListenerContainer() {
          private long deactivatedTill = 0;

          @Override
          protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
            if (deactivatedTill < System.currentTimeMillis()) {
              return receiveFromConsumer(consumer, getReceiveTimeout());
            }
            logger.info("Disabled due to failure :(");
            return null;
          }

          @Override
          protected void doInvokeListener(MessageListener listener, Message message)
              throws JMSException {
            try {
              super.doInvokeListener(listener, message);
            } catch (Exception e) {
              handleException(message);
              throw e;
            }
          }

          private long getDelay(int retryCount) {
            if (retryCount <= 1) {
              return 20;
            }
            return (long) (20 * Math.pow(2, retryCount));
          }

          private void handleException(Message msg) throws JMSException {
            if (msg.propertyExists("JMSXDeliveryCount")) {
              int retryCount = msg.getIntProperty("JMSXDeliveryCount");
              deactivatedTill = System.currentTimeMillis() + getDelay(retryCount);
            }
          }

          @Override
          protected void doInvokeListener(SessionAwareMessageListener listener, Session session,
              Message message)
              throws JMSException {
            try {
              super.doInvokeListener(listener, session, message);
            } catch (Exception e) {
              handleException(message);
              throw e;
            }
          }
        };
      }
    };
    // This provides all boot's default to this factory, including the message converter
    configurer.configure(factory, connectionFactory);
    // You could still override some of Boot's default if necessary.
    return factory;
  }
}