1. 程式人生 > >Spring Cloud Stream消費失敗後的處理策略(一):自動重試

Spring Cloud Stream消費失敗後的處理策略(一):自動重試

之前寫了幾篇關於Spring Cloud Stream使用中的常見問題,比如:

下面幾天就集中來詳細聊聊,當訊息消費失敗之後該如何處理的幾種方式。不過不論哪種方式,都需要與具體業務結合,解決不同業務場景可能出現的問題。

今天第一節,介紹一下Spring Cloud Stream中預設就已經配置了的一個異常解決方案:重試!

應用場景

依然要明確一點,任何解決方案都要結合具體的業務實現來確定,不要有了錘子看什麼問題都是釘子。那麼重試可以解決什麼問題呢?由於重試的基礎邏輯並不會改變,所以通常重試只能解決因環境不穩定等外在因素導致的失敗情況,比如:當我們接收到某個訊息之後,需要呼叫一個外部的Web Service做一些事情,這個時候如果與外部系統的網路出現了抖動,導致呼叫失敗而丟擲異常。這個時候,通過重試訊息消費的具體邏輯,可能在下一次呼叫的時候,就能完成整合業務動作,從而解決剛才所述的問題。

動手試試

先通過一個小例子來看看Spring Cloud Stream預設的重試機制是如何運作的。之前在如何消費自己生產的訊息一文中的例子,我們可以繼續沿用,或者也可以精簡一些,都寫到一個主類中,比如下面這樣:

@EnableBinding(TestApplication.TestTopic.class)
@SpringBootApplication
public class TestApplication {

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

    @RestController
    static class TestController {

        @Autowired
        private TestTopic testTopic;

        /**
         * 訊息生產介面
         * @param message
         * @return
         */
        @GetMapping("/sendMessage")
        public String messageWithMQ(@RequestParam String message) {
            testTopic.output().send(MessageBuilder.withPayload(message).build());
            return "ok";
        }

    }

    /**
     * 訊息消費邏輯
     */
    @Slf4j
    @Component
    static class TestListener {

        @StreamListener(TestTopic.INPUT)
        public void receive(String payload) {
            log.info("Received: " + payload);
            throw new RuntimeException("Message consumer failed!");
        }

    }

    interface TestTopic {

        String OUTPUT = "example-topic-output";
        String INPUT = "example-topic-input";

        @Output(OUTPUT)
        MessageChannel output();

        @Input(INPUT)
        SubscribableChannel input();

    }

}

內容很簡單,既包含了訊息的生產,也包含了訊息消費。與之前例子不同的就是在訊息消費邏輯中,主動的丟擲了一個異常來模擬訊息的消費失敗。

在啟動應用之前,還要記得配置一下輸入輸出通道對應的物理目標(exchange或topic名),比如:

spring.cloud.stream.bindings.example-topic-input.destination=test-topic
spring.cloud.stream.bindings.example-topic-output.destination=test-topic

完成了上面配置之後,就可以啟動應用,並嘗試訪問localhost:8080/sendMessage?message=hello

介面來發送一個訊息到MQ中了。此時可以看到類似下面的日誌:

2018-12-10 11:20:21.345  INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener  : Received: hello
2018-12-10 11:20:22.350  INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener  : Received: hello
2018-12-10 11:20:24.354  INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener  : Received: hello
2018-12-10 11:20:54.651 ERROR 30499 --- [w2p2yKethOsqg-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Exception thrown while invoking com.didispace.stream.TestApplication$TestListener#receive[1 args]; nested exception is java.lang.RuntimeException: Message consumer failed!, failedMessage=GenericMessage [payload=byte[5], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=test-topic, amqp_receivedExchange=test-topic, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=test-topic.anonymous.EuqBJu66Qw2p2yKethOsqg, amqp_redelivered=false, id=a89adf96-7de2-f29d-20b6-2fcb0c64cd8c, amqp_consumerTag=amq.ctag-XFy6vXU2w4RB_NRBzImWTA, contentType=application/json, timestamp=1544412051638}]
	at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:63)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:158)
	at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)
	at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1100(AmqpInboundChannelAdapter.java:60)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:214)
	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
	at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:211)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1414)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1337)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1324)
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1303)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77)
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Message consumer failed!
	at com.didispace.stream.TestApplication$TestListener.receive(TestApplication.java:65)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:181)
	at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:114)
	at org.springframework.cloud.stream.binding.StreamListenerMessageHandler.handleRequestMessage(StreamListenerMessageHandler.java:55)
	... 27 more

從日誌中可以看到,一共輸出了三次Received: hello,也就是說訊息消費邏輯執行了3次,然後丟擲了最終執行失敗的異常。

設定重複次數

預設情況下Spring Cloud Stream會重試3次,我們也可以通過配置的方式修改這個預設配置,比如下面的配置可以將重試次數調整為1次:

spring.cloud.stream.bindings.example-topic-input.consumer.max-attempts=1

對於一些純內部計算邏輯,不需要依賴外部環境,如果出錯通常是程式碼邏輯錯誤的情況下,不論我們如何重試都會繼續錯誤的業務邏輯可以將該引數設定為0,避免不必要的重試影響訊息處理的速度。

深入思考

完成了上面的基礎嘗試之後,再思考下面兩個問題:

問題一:如果在重試過程中訊息處理成功了,還會有異常資訊嗎?

答案是不會。因為重試過程是訊息處理的一個整體,如果某一次重試成功了,會任務對所收到訊息的消費成功了。

這個問題可以在上述例子中做一些小改動來驗證,比如:

@Slf4j
@Component
static class TestListener {

    int counter = 1;

    @StreamListener(TestTopic.INPUT)
    public void receive(String payload) {
        log.info("Received: " + payload + ", " + counter);
        if (counter == 3) {
            counter = 1;
            return;
        } else {
            counter++;
            throw new RuntimeException("Message consumer failed!");
        }
    }

}

通過加入一個計數器,當重試是第3次的時候,不丟擲異常來模擬消費邏輯處理成功了。此時重新執行程式,並呼叫介面localhost:8080/sendMessage?message=hello,可以獲得如下日誌結果,並沒有異常打印出來。

2018-12-10 16:07:38.390  INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener  : Received: hello, 1
2018-12-10 16:07:39.398  INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener  : Received: hello, 2
2018-12-10 16:07:41.402  INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener  : Received: hello, 3

也就是,雖然前兩次消費丟擲了異常,但是並不影響最終的結果,也不會列印中間過程的異常,避免了對日誌告警產生誤報等問題。

問題二:如果重試都失敗之後應該怎麼辦呢?

如果訊息在重試了還是失敗之後,目前的配置唯一能做的就是將異常資訊記錄下來,進行告警。由於日誌中有訊息的訊息資訊描述,所以應用維護者可以根據這些資訊來做一些補救措施。

當然,這樣的做法顯然不是最好的,因為太過麻煩。那麼怎麼做才好呢?且聽下回分解!

程式碼示例

本文示例讀者可以通過檢視下面倉庫的中的stream-exception-handler-1專案:

如果您對這些感興趣,歡迎star、follow、收藏、轉發給予支援!

以下專題教程也許您會有興趣