spring - 获取原因 :null in property dlqDeliveryFailureCause

Question

我正在尝试为系统设置死信队列监控。到目前为止，当消费者的消息消费失败时，我可以毫无问题地将它扔进 DLQ 队列。现在我在弄清它失败的原因时遇到了一些麻烦；

目前我得到以下内容

java.lang.Throwable: Delivery[2] exceeds redelivery policy imit:RedeliveryPolicy 
  {destination = queue://*, 
   collisionAvoidanceFactor = 0.15, 
   maximumRedeliveries = 1, 
   maximumRedeliveryDelay = -1, 
   initialRedeliveryDelay = 10000, 
   useCollisionAvoidance = false, 
   useExponentialBackOff = true, 
   backOffMultiplier = 5.0, 
   redeliveryDelay = 10000, 
   preDispatchCheck = true}, 
   cause:null

我不知道为什么 cause 返回为 null。我将 Spring 与 ActiveMQ 一起使用。我正在使用 DefaultJmsListenerContainerFactory，它创建一个 DefaultMessageListenerContainer。我希望 cause 充满发生在我的 consumer 上的异常，但我无法让它工作。显然 Spring 上有些东西没有正确冒出异常，但我不确定它是什么。我正在使用 spring-jms:4.3.10。非常感谢您的帮助。

Answer 1

我正在使用 spring-boot-starter-activemq:2.2.2.RELEASE (spring-jms:5.2.2, activemq-client-5.15 .11) 和我有同样的行为。

(链接指向我使用的版本)

添加了回滚原因here对于 POSION_ACK_TYPE(原文如此!)。它的assignment MessageDispatch 仅发生在 one place 中: 在处理 RuntimeException 时 如果有 javax.jms.MessageListener 已注册。

不幸的是(对于这种特殊情况)，Spring 没有注册一个，因为它更喜欢处理自己的层次结构。所以，长话短说，开箱即用的 Spring 没有机会实现它。

但是，我设法编写了一种 hack-ish 方式来访问所处理的 MessageDispatch 实例，注入(inject)异常作为回滚原因，并且它有效!

package com.example;

import org.springframework.jms.listener.DefaultMessageListenerContainer;

import javax.jms.*;

public class MyJmsMessageListenerContainer extends DefaultMessageListenerContainer {

    private final MessageDeliveryFailureCauseEnricher messageDeliveryFailureCauseEnricher = new MessageDeliveryFailureCauseEnricher();

    private MessageConsumer messageConsumer; // Keep for later introspection

    @Override
    protected MessageConsumer createConsumer(Session session, Destination destination) throws JMSException {
        this.messageConsumer = super.createConsumer(session, destination);
        return this.messageConsumer;
    }

    @Override
    protected void invokeListener(Session session, Message message) throws JMSException {
        try {
            super.invokeListener(session, message);
        } catch (Throwable throwable) {
            messageDeliveryFailureCauseEnricher.enrich(throwable, this.messageConsumer);
            throw throwable;
        }
    }
}

注意:不要通过覆盖 protected void handleListenerException(Throwable ex) 方法来处理 Throwable，因为那时 中已经发生了一些清理>ActiveMQMessageConsumer 实例。

package com.example;

import org.apache.activemq.ActiveMQMessageConsumer;
import org.apache.activemq.command.MessageDispatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ReflectionUtils;

import javax.jms.MessageConsumer;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

class MessageDeliveryFailureCauseEnricher {
    private static final Logger logger = LoggerFactory.getLogger(MessageDeliveryFailureCauseEnricher.class);

    private final Map<Class<?>, Field> accessorFields = new HashMap<>();
    private final Field targetField;

    public MessageDeliveryFailureCauseEnricher() {
        this.targetField = register(ActiveMQMessageConsumer.class, "deliveredMessages");
        // Your mileage may vary; here is mine:
        register("brave.jms.TracingMessageConsumer", "delegate");
        register("org.springframework.jms.connection.CachedMessageConsumer", "target");
    }

    private Field register(String className, String fieldName) {
        Field result = null;
        if (className == null) {
            logger.warn("Can't register a field from a missing class name");
        } else {
            try {
                Class<?> clazz = Class.forName(className);
                result = register(clazz, fieldName);
            } catch (ClassNotFoundException e) {
                logger.warn("Class not found on classpath: {}", className);
            }
        }
        return result;
    }

    private Field register(Class<?> clazz, String fieldName) {
        Field result = null;
        if (fieldName == null) {
            logger.warn("Can't register a missing class field name");
        } else {
            Field field = ReflectionUtils.findField(clazz, fieldName);
            if (field != null) {
                ReflectionUtils.makeAccessible(field);
                accessorFields.put(clazz, field);
            }
            result = field;
        }
        return result;
    }

    void enrich(Throwable throwable, MessageConsumer messageConsumer) {
        if (throwable != null) {
            if (messageConsumer == null) {
                logger.error("Can't enrich the MessageDispatch with rollback cause '{}' if no MessageConsumer is provided", throwable.getMessage());
            } else {
                LinkedList<MessageDispatch> deliveredMessages = lookupFrom(messageConsumer);
                if (deliveredMessages != null && !deliveredMessages.isEmpty()) {
                    deliveredMessages.getLast().setRollbackCause(throwable); // Might cause problems if we prefetch more than 1 message
                }
            }
        }
    }

    private LinkedList<MessageDispatch> lookupFrom(Object object) {
        LinkedList<MessageDispatch> result = null;
        if (object != null) {
            Field field = accessorFields.get(object.getClass());
            if (field != null) {
                Object fieldValue = ReflectionUtils.getField(field, object);
                if (fieldValue != null) {
                    if (targetField == field) {
                        result = (LinkedList<MessageDispatch>) fieldValue;
                    } else {
                        result = lookupFrom(fieldValue);
                    }
                }
            }
        }

        return result;
    }
}

魔法发生在第二堂课:

1. 在施工期间，我们让一些私有(private)领域可访问。
2. 当捕获到 Throwable 时，我们遍历这些字段以得到适当的 MessageDispatch 实例(请注意，如果您预取了超过 1 条消息)，并将其注入(inject)throwable 我们希望成为 dlqDeliveryFailureCause JMS 属性的一部分。

经过数小时的调试(感谢 OSS!)以及多次试验和错误，我今天下午精心设计了这个解决方案。它有效，但我觉得它更像是一种 hack，而不是真正可靠的解决方案。考虑到这一点，我尽最大努力避免副作用，所以最坏的情况就是在以死信队列结尾的消息中找不到原始 Throwable 的踪迹。

如果我在某处遗漏了要点，我很乐意了解更多相关信息。