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

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

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

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

閱讀更多 Java小可愛 的文章

關鍵字: 錘子 技術 重試