JMS 在不阻塞 onMessage() 的情况下配置退避/重试

Question

javax.JMS 版本 2.0.1

提供者：ibm.mq v9.0

框架：Java Spring 启动

据我所知，onMessage() 是异步的。我正在成功重试消息发送。但是，消息的重新发送会在消息失败后立即发生。理想情况下，我希望重试以滑动窗口样式发生，例如。 20 秒后第一次重试，40 秒后第二次重试，以此类推。

如果没有一个 Thread.Sleep()，我想这会阻塞整个 Java 线程，而这根本不是我想要的，我该如何实现呢？

代码是这样的

final int TIME_TO_WAIT = 20;

public void onMessage(Message , message)
{
   :
   :
   int t  =   message.getIntProperty("JMSXDeliveryCount");
   if(t > 1)
   {
     // Figure out a way to wait for (TIME_TO_WAIT * t)
   
   }

}
catch(Exception e)
{
    // Do some logging/cleanup etc.
    throw new RunimeException(e);// this causes a message retry
}

Answer 1

我建议您在重试逻辑中使用指数退避，但您需要使用 Delivery Delayfeature。

定义一个自定义 JmsTemplate 将使用消息中的延迟属性，您还应该在消息属性中添加重试计数，以便您可以根据需要延迟，例如 20、40、80、160 等

public class DelayedJmsTemplate extends JmsTemplate {
    public static String DELAY_PROPERTY_NAME = "deliveryDelay";

    @Override
    protected void doSend(MessageProducer producer, Message message) throws JMSException {
        long delay = -1;
        if (message.propertyExists(DELAY_PROPERTY_NAME)) {
            delay = message.getLongProperty(DELAY_PROPERTY_NAME);
        }
        if (delay >= 0) {
            producer.setDeliveryDelay(delay);
        }
        if (isExplicitQosEnabled()) {
            producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
        } else {
            producer.send(message);
        }
    }
}

定义组件，将具有重新排队消息的能力，您可以在基本消息侦听器中定义此接口。 handleException 方法应该完成入队和计算延迟等所有任务。您可能并不总是对入队感兴趣，在某些情况下，您也会丢弃消息。

您可以在此处看到类似的后处理逻辑 https://github.com/sonus21/rqueue/blob/4c9c5c88f02e5cf0ac4b16129fe5b880411d7afc/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/PostProcessingHandler.java

@Component
@Sl4j
public class MessageListener {
    private final JmsTemplate jmsTemplate;

    @Autowired
    public MessageListener(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }
    
    @JmsListener(destination = "myDestination")
    public void onMessage(Message message) throws JMSException {
        try {
            // do something
        } catch (Exception e) {
            handleException("myDestination", message, e);
        }
    }

    // Decide whether the message should be ignored due to many retries etc
    private boolean shouldBeIgnored(String destination, Message message) {
        return false;
    }

    // add logic to compute delay
    private long getDelay(String destination, Message message, int deliveryCount) {
        return 100L;
    }

    private void handleException(String destination, Message message, Exception e) throws JMSException {
        if (shouldBeIgnored(destination, message)) {
            log.info("destination: {}, message: {} is ignored ", destination, message, e);
            return;
        }
        if (message.propertyExists("JMSXDeliveryCount")) {
            int t = message.getIntProperty("JMSXDeliveryCount");
            long delay = getDelay(destination, message, t + 1);
            message.setLongProperty(DELAY_PROPERTY_NAME, delay);
            message.setIntProperty("JMSXDeliveryCount", t + 1);
            jmsTemplate.send(destination, session -> message);
        } else {
            // no delivery count, is this the first message or should be ignored?
        }
    }
}