重試關鍵業務邏輯

在某些情境中，您可能想要重試對於應用程式至關重要的部分業務邏輯。可能是對關係資料庫的外部呼叫，或是從 Kafka Streams 處理器調用 REST 端點。這些呼叫可能會因各種原因而失敗，例如網路問題或遠端服務不可用。通常，如果您可以再次嘗試，這些失敗可能會自行解決。預設情況下，Kafka Streams 綁定器會為所有輸入綁定建立 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 提供，則它會優先於任何輸入綁定層級的重試範本組態屬性。

一旦從綁定注入 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;
    ));

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