如何在 Spring Cloud RabbitMQ 中拦截重新发布到 DLQ 的消息？

Question

我想拦截重试限制用完后重新发布到 DLQ 的消息，我的最终目标是从这些消息中消除 x-exception-stacktrace 标头。

配置：

spring:
  application:
    name: sandbox
  cloud:
    function:
      definition: rabbitTest1Input
    stream:
      binders:
        rabbitTestBinder1:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: localhost:55015
                username: guest
                password: guest
                virtual-host: test
 
      bindings:
        rabbitTest1Input-in-0:
          binder: rabbitTestBinder1
          consumer:
            max-attempts: 3
          destination: ex1
          group: q1
      rabbit:
        bindings:
          rabbitTest1Input-in-0:
            consumer:
              autoBindDlq: true
              bind-queue: true
              binding-routing-key: q1key
              deadLetterExchange: ex1-DLX
              dlqDeadLetterExchange: ex1
              dlqDeadLetterRoutingKey: q1key_dlq
              dlqTtl: 180000
              prefetch: 5
              queue-name-group-only: true
              republishToDlq: true
              requeueRejected: false
              ttl: 86400000

@Configuration
class ConsumerConfig {

    companion object : KLogging()

    @Bean
    fun rabbitTest1Input(): Consumer<Message<String>> {
        return Consumer {
            logger.info("Received from test1 queue: ${it.payload}")
            throw AmqpRejectAndDontRequeueException("FAILED")  // force republishing to DLQ after N retries
        }
    }
}

首先我尝试注册@GlobalChannelInterceptor（如here），但由于RabbitMessageChannelBinder 使用其自己的私有RabbitTemplate 实例（未自动连接）进行重新发布（请参阅#getErrorMessageHandler），它不会被拦截。

然后我尝试通过丢弃与x-exception-stacktrace相关的代码来扩展RabbitMessageChannelBinder类，然后将此扩展声明为bean：

/**
 * Forked from {@link org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder} with the goal
 * to eliminate {@link RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE} header from messages republished to DLQ
 */
class RabbitMessageChannelBinderWithNoStacktraceRepublished 
    : RabbitMessageChannelBinder(...)

// and then

@Configuration
@Import(
    RabbitAutoConfiguration::class,
    RabbitServiceAutoConfiguration::class,
    RabbitMessageChannelBinderConfiguration::class,
    PropertyPlaceholderAutoConfiguration::class,
)
@EnableConfigurationProperties(
    RabbitProperties::class,
    RabbitBinderConfigurationProperties::class,
    RabbitExtendedBindingProperties::class
)
class RabbitConfig {

    @Bean
    @Primary
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    @Order(Ordered.HIGHEST_PRECEDENCE)
    fun customRabbitMessageChannelBinder(
        appCtx: ConfigurableApplicationContext,
        ... // required injections
    ): RabbitMessageChannelBinder {

        // remove the original (auto-configured) bean. Explanation is after the code snippet
        val registry = appCtx.autowireCapableBeanFactory as BeanDefinitionRegistry
        registry.removeBeanDefinition("rabbitMessageChannelBinder")

        // ... and replace it with custom binder. It's initialized absolutely the same way as original bean, but is of forked class
        return RabbitMessageChannelBinderWithNoStacktraceRepublished(...)
    }
}

但在这种情况下，我的频道绑定器不尊重 YAML 属性（例如 addresses: localhost:55015）并使用默认值（例如 localhost:5672）

INFO  o.s.a.r.c.CachingConnectionFactory - Attempting to connect to: [localhost:5672]
INFO  o.s.a.r.l.SimpleMessageListenerContainer - Broker not available; cannot force queue declarations during start: java.net.ConnectException: Connection refused

另一方面，如果我不从 Spring 上下文中删除原始活页夹，我会收到以下错误：

Caused by: java.lang.IllegalStateException: Multiple binders are available, however neither default nor per-destination binder name is provided. Available binders are [rabbitMessageChannelBinder, customRabbitMessageChannelBinder]
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinder(DefaultBinderFactory.java:145)

谁能给我一个提示如何解决这个问题？

P.S.我使用 Spring Cloud Stream 3.1.6 和 Spring Boot 2.6.6

Answer 1

1. 禁用活页夹重试/DLQ 配置（maxAttempts=1、republishToDlq=false 和其他 dlq 相关属性）。
2. 添加 ListenerContainerCustomizer 以向建议链添加自定义重试建议，并使用自定义死信发布恢复器。
3. 使用 Queue @Bean 手动配置 DLQ。

@SpringBootApplication
public class So72871662Application {

    public static void main(String[] args) {
        SpringApplication.run(So72871662Application.class, args);
    }

    @Bean
    public Consumer<String> input() {
        return str -> {
            System.out.println();
            throw new RuntimeException("test");
        };
    }

    @Bean
    ListenerContainerCustomizer<MessageListenerContainer> customizer(RetryOperationsInterceptor retry) {
        return (cont, dest, grp) -> {
            ((AbstractMessageListenerContainer) cont).setAdviceChain(retry);
        };
    }

    @Bean
    RetryOperationsInterceptor interceptor(MessageRecoverer recoverer) {
        return RetryInterceptorBuilder.stateless()
                .maxAttempts(3)
                .backOffOptions(3_000L, 2.0, 10_000L)
                .recoverer(recoverer)
                .build();
    }

    @Bean
    MessageRecoverer recoverer(RabbitTemplate template) {
        return new RepublishMessageRecoverer(template, "DLX", "errors") {

            @Override
            protected void doSend(@Nullable
            String exchange, String routingKey, Message message) {

                message.getMessageProperties().getHeaders().remove(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE);
                super.doSend(exchange, routingKey, message);
            }

        };
    }

    @Bean
    FanoutExchange dlx() {
        return new FanoutExchange("DLX");
    }

    @Bean
    Queue dlq() {
        return new Queue("errors");
    }

    @Bean
    Binding dlqb() {
        return BindingBuilder.bind(dlq()).to(dlx());
    }

}