死信主题处理

启用 DLQ

要启用 DLQ，基于 Kafka Binder 的应用程序必须通过属性 `spring.cloud.stream.bindings.<binding-name>.group` 提供消费者组。匿名消费者组（即应用程序没有显式提供组）无法启用 DLQ 功能。

当应用程序想要将错误记录发送到 DLQ 主题时，该应用程序必须启用 DLQ 功能，因为默认情况下这是未启用的。要启用 DLQ，必须将属性 `spring.cloud.stream.kafka.bindings.<binding-name>.consumer.enable-dlq` 设置为 true。

启用 DLQ 后，在处理发生错误并且基于 `spring.cloud.stream.bindings.<binding-name>.consumer.max-attempts` 属性用尽所有重试后，该记录将被发送到 DLQ 主题。

默认情况下，`max-attempts` 属性设置为 3。当 `max-attempts` 属性大于 `1` 且启用了 dlq 时，您会看到重试遵循 `max-attempts` 属性。当未启用 dlq（这是默认值）时，`max-attempts` 属性对重试处理方式没有任何影响。在这种情况下，重试将回退到 Apache Kafka 中 Spring 的容器默认值，即 `10` 次重试。如果应用程序想要在禁用 DLQ 时完全禁用重试，则将 `max-attempts` 属性设置为 `1` 将不起作用。在这种情况下，要完全禁用重试，需要提供一个 `ListenerContainerCustomizer`，然后使用适当的 `Backoff` 设置。这是一个示例。

@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
	return (container, destinationName, group) -> {
		var commonErrorHandler = new DefaultErrorHandler(new FixedBackOff(0L, 0l));
		container.setCommonErrorHandler(commonErrorHandler);
	};
}

这样，默认的容器行为将被禁用，并且不会尝试任何重试。如上所述，启用 DLQ 时，Binder 设置将具有优先级。

处理死信主题中的记录

由于框架无法预测用户如何处理死信消息，因此它不提供任何标准机制来处理这些消息。如果死信的原因是暂时的，您可能希望将消息路由回原始主题。但是，如果问题是永久性的，则可能会导致无限循环。本主题中的 Spring Boot 示例应用程序演示了如何将这些消息路由回原始主题，但它会在三次尝试后将它们移动到“停车场”主题。该应用程序是另一个 spring-cloud-stream 应用程序，它从死信主题读取消息。如果 5 秒内没有收到消息，它将退出。

这些示例假设原始目标是so8400out，使用者组是so8400。

有几种策略可供考虑

  • 考虑仅在主应用程序未运行时运行重新路由。否则，瞬态错误的重试次数很快就会用完。

  • 或者，使用两阶段方法：使用此应用程序路由到第三个主题，然后使用另一个应用程序从那里路由回主主题。

以下代码清单显示了示例应用程序

application.properties
spring.cloud.stream.bindings.input.group=so8400replay
spring.cloud.stream.bindings.input.destination=error.so8400out.so8400

spring.cloud.stream.bindings.output.destination=so8400out

spring.cloud.stream.bindings.parkingLot.destination=so8400in.parkingLot

spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=earliest

spring.cloud.stream.kafka.binder.headers=x-retries
应用程序
@SpringBootApplication
public class ReRouteDlqKApplication implements CommandLineRunner {

    private static final String X_RETRIES_HEADER = "x-retries";

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

    private final AtomicInteger processed = new AtomicInteger();

    @Autowired
    private StreamBridge streamBridge;

    @Bean
    public Function<Message<?>, Message<?>> reRoute() {
        return failed -> {
            processed.incrementAndGet();
            Integer retries = failed.getHeaders().get(X_RETRIES_HEADER, Integer.class);
            if (retries == null) {
                System.out.println("First retry for " + failed);
                return MessageBuilder.fromMessage(failed)
                        .setHeader(X_RETRIES_HEADER, 1)
                        .setHeader(BinderHeaders.PARTITION_OVERRIDE,
                                failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
                        .build();
            }
            else if (retries < 3) {
                System.out.println("Another retry for " + failed);
                return MessageBuilder.fromMessage(failed)
                        .setHeader(X_RETRIES_HEADER, retries + 1)
                        .setHeader(BinderHeaders.PARTITION_OVERRIDE,
                                failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
                        .build();
            }
            else {
                System.out.println("Retries exhausted for " + failed);
                streamBridge.send("parkingLot", MessageBuilder.fromMessage(failed)
                    .setHeader(BinderHeaders.PARTITION_OVERRIDE,
                            failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
                    .build());
            }
            return null;
        };
    }

    @Override
    public void run(String... args) throws Exception {
        while (true) {
            int count = this.processed.get();
            Thread.sleep(5000);
            if (count == this.processed.get()) {
                System.out.println("Idle, exiting");
                return;
            }
        }
    }
}