Spring Cloud Stream消費失敗後的處理策略（一）：自動重試
阿新 • • 發佈：2018-12-13
之前寫了幾篇關於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
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、收藏、轉發給予支援！