重試關鍵業務邏輯

在某些場景下，你可能需要重試應用程式中關鍵的業務邏輯部分。這可能是對關係型資料庫的外部呼叫，或是從 Kafka Streams 處理器呼叫 REST 端點。這些呼叫可能因各種原因失敗，例如網路問題或遠端服務不可用。通常情況下，如果再次嘗試，這些失敗可能會自行解決。預設情況下，Kafka Streams binder 會為所有輸入繫結建立 RetryTemplate bean。

如果函式具有以下簽名：

@Bean
public java.util.function.Consumer<KStream<Object, String>> process()

使用預設繫結名稱時，RetryTemplate 將註冊為 process-in-0-RetryTemplate。這遵循繫結名稱 (process-in-0) 後面跟字面量 -RetryTemplate 的約定。對於多個輸入繫結，每個繫結都會有一個單獨的 RetryTemplate bean 可用。如果應用程式中存在自定義的 RetryTemplate bean，並透過 spring.cloud.stream.bindings.<binding-name>.consumer.retryTemplateName 提供，則該自定義 bean 將優先於任何輸入繫結級別的重試模板配置屬性。

一旦繫結中的 RetryTemplate 被注入到應用程式中，就可以用來重試應用程式中的任何關鍵部分。示例如下：

@Bean
public java.util.function.Consumer<KStream<Object, String>> process(@Lazy @Qualifier("process-in-0-RetryTemplate") RetryTemplate retryTemplate) {

    return input -> input
            .process(() -> new Processor<Object, String>() {
                @Override
                public void init(ProcessorContext processorContext) {
                }

                @Override
                public void process(Object o, String s) {
                    retryTemplate.execute(context -> {
                       //Critical business logic goes here.
                    });
                }

                @Override
                public void close() {
                }
            });
}

或者，你可以使用自定義的 RetryTemplate，如下所示：

@EnableAutoConfiguration
public static class CustomRetryTemplateApp {

    @Bean
    @StreamRetryTemplate
    RetryTemplate fooRetryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        RetryPolicy retryPolicy = new SimpleRetryPolicy(4);
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1);

        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }

    @Bean
    public java.util.function.Consumer<KStream<Object, String>> process() {

        return input -> input
                .process(() -> new Processor<Object, String>() {
                    @Override
                    public void init(ProcessorContext processorContext) {
                    }

                    @Override
                    public void process(Object o, String s) {
                        fooRetryTemplate().execute(context -> {
                           //Critical business logic goes here.
                        });

                    }

                    @Override
                    public void close() {
                    }
                });
    }
}

注意，當重試次數耗盡時，預設情況下會丟擲最後一個異常，導致處理器終止。如果你希望處理異常並繼續處理，可以將一個 RecoveryCallback 新增到 execute 方法中。示例如下：

retryTemplate.execute(context -> {
    //Critical business logic goes here.
    }, context -> {
       //Recovery logic goes here.
       return null;
    ));

有關 Spring Retry 專案的 RetryTemplate、重試策略 (retry policies)、退避策略 (backoff policies) 等更多資訊，請參閱該專案。