Spring Cloud Stream Rabbit自定义错误处理器与republishToDlq协同问题

配置了自定义错误处理器且republishToDlq为true，但死信队列（DLQ）中的消息处于被拒绝状态而非重发布状态，同时缺少x-exception-stacktrace等异常信息。希望同时使用自定义错误处理器与republishToDlq（需获取x-exception-stacktrace），该如何实现？或者是否应改用@GlobalChannelInterceptor(patterns = "*error*")拦截器替代自定义错误处理器（用于发送邮件、审计等操作）？

现有配置 #

spring:
  cloud:
    stream:
      default.error-handler-definition: myErrorHandler
      function.definition: personSave

      bindings:
        personSave-in-0:
          destination: personSaveRequest
          group: personSaveRequestGroup
          content-type: application/json

      rabbit:
        bindings:
          personSave-in-0:
            consumer:
              autoBindDlq: true
              deadLetterExchange: nest.dlx
              deadLetterQueueName: nest.dlq

自定义错误处理器代码 #

@Bean
public Consumer<ErrorMessage> myErrorHandler() {
        return errorMessage -> {
                        // 发送邮件、审计等操作
            log.error("Some error: {}", errorMessage);
            ReflectionUtils.rethrowRuntimeException(errorMessage.getPayload());
    };
}

问题根源 #

当前自定义错误处理器中调用ReflectionUtils.rethrowRuntimeException(errorMessage.getPayload())，会导致异常被重新抛出，最终由RabbitMQ默认机制将消息拒绝并移至死信队列，而非通过republishToDlq逻辑完成重发布，因此不会添加x-exception-stacktrace等异常头信息。

实现自定义错误处理器+republishToDlq的正确方式 #

要同时实现自定义逻辑（邮件、审计）和带异常信息的DLQ重发布，需调整自定义错误处理器的逻辑，不要直接抛出异常，而是借助Spring Cloud Stream内置的RepublishMessageRecoverer完成重发布：

1. 移除配置中的default.error-handler-definition，手动配置错误处理器结合消息恢复器：

@Bean
public ErrorHandler myErrorHandler(RepublishMessageRecoverer recoverer) {
    return t -> {
        // 执行自定义逻辑：发送邮件、审计
        log.error("处理异常，执行自定义操作", t);
        // 委托RepublishMessageRecoverer完成带异常头的DLQ重发布
        recoverer.recover(((ListenerExecutionFailedException) t).getFailedMessage(), t);
    };
}

@Bean
public RepublishMessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
    return new RepublishMessageRecoverer(rabbitTemplate, "nest.dlx", "nest.dlq");
}

2. 调整Rabbit消费者配置，显式开启republishToDlq（默认值为false）：

spring:
  cloud:
    stream:
      rabbit:
        bindings:
          personSave-in-0:
            consumer:
              autoBindDlq: true
              deadLetterExchange: nest.dlx
              deadLetterQueueName: nest.dlq
              republishToDlq: true

关于@GlobalChannelInterceptor的替代方案 #

如果不想调整错误处理器逻辑，改用@GlobalChannelInterceptor拦截*error*通道也是可行的。这种方式可以在消息进入错误通道时执行自定义操作，同时不影响内置的republishToDlq逻辑：

@Component
public class ErrorChannelInterceptor implements ChannelInterceptor {

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        if (channel.getName().endsWith("error")) {
            ErrorMessage errorMessage = (ErrorMessage) message;
            // 执行自定义逻辑：发送邮件、审计
            log.error("捕获错误消息，执行自定义操作: {}", errorMessage);
        }
        return message;
    }
}

该方案无需自定义错误处理器，只需保留republishToDlq: true的配置，内置重发布逻辑会自动添加x-exception-stacktrace等异常头信息，拦截器负责完成自定义操作。

内容的提问来源于stack exchange，提问作者ltom74