死信主題處理

啟用 DLQ

要啟用 DLQ，基於 Kafka 繫結的應用程式必須透過屬性 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 屬性對重試的處理方式沒有任何影響。在這種情況下，重試將回退到 Spring for Apache Kafka 中的容器預設值，即 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 時，繫結器設定將具有優先權。

處理死信主題中的記錄

由於框架無法預測使用者希望如何處理死信訊息，因此它不提供任何標準機制來處理它們。如果死信的原因是暫時的，您可能希望將訊息路由回原始主題。但是，如果問題是永久性的，可能會導致無限迴圈。本主題中的 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;
            }
        }
    }
}