油金地 做网站,代做一个网站多少钱,网站建设出错1004,中国建设银行个人查询余额之前写了几篇关于Spring Cloud Stream使用中的常见问题#xff0c;比如#xff1a;
如何处理消息重复消费如何消费自己生产的消息
下面几天就集中来详细聊聊#xff0c;当消息消费失败之后该如何处理的几种方式。不过不论哪种方式#xff0c;都需要与具体业务结合#xf…之前写了几篇关于Spring Cloud Stream使用中的常见问题比如
如何处理消息重复消费如何消费自己生产的消息
下面几天就集中来详细聊聊当消息消费失败之后该如何处理的几种方式。不过不论哪种方式都需要与具体业务结合解决不同业务场景可能出现的问题。
今天第一节介绍一下Spring Cloud Stream中默认就已经配置了的一个异常解决方案重试
应用场景
依然要明确一点任何解决方案都要结合具体的业务实现来确定不要有了锤子看什么问题都是钉子。那么重试可以解决什么问题呢由于重试的基础逻辑并不会改变所以通常重试只能解决因环境不稳定等外在因素导致的失败情况比如当我们接收到某个消息之后需要调用一个外部的Web Service做一些事情这个时候如果与外部系统的网络出现了抖动导致调用失败而抛出异常。这个时候通过重试消息消费的具体逻辑可能在下一次调用的时候就能完成整合业务动作从而解决刚才所述的问题。
动手试试
先通过一个小例子来看看Spring Cloud Stream默认的重试机制是如何运作的。之前在如何消费自己生产的消息一文中的例子我们可以继续沿用或者也可以精简一些都写到一个主类中比如下面这样
EnableBinding(TestApplication.TestTopic.class)SpringBootApplicationpublic 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.destinationtest-topicspring.cloud.stream.bindings.example-topic-output.destinationtest-topic完成了上面配置之后就可以启动应用并尝试访问localhost:8080/sendMessage?messagehello接口来发送一个消息到MQ中了。此时可以看到类似下面的日志
2018-12-10 11:20:21.345 INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener : Received: hello2018-12-10 11:20:22.350 INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener : Received: hello2018-12-10 11:20:24.354 INFO 30499 --- [w2p2yKethOsqg-1] c.d.stream.TestApplication$TestListener : Received: hello2018-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!, failedMessageGenericMessage [payloadbyte[5], headers{amqp_receivedDeliveryModePERSISTENT, amqp_receivedRoutingKeytest-topic, amqp_receivedExchangetest-topic, amqp_deliveryTag2, deliveryAttempt3, amqp_consumerQueuetest-topic.anonymous.EuqBJu66Qw2p2yKethOsqg, amqp_redeliveredfalse, ida89adf96-7de2-f29d-20b6-2fcb0c64cd8c, amqp_consumerTagamq.ctag-XFy6vXU2w4RB_NRBzImWTA, contentTypeapplication/json, timestamp1544412051638}] 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-attempts1对于一些纯内部计算逻辑不需要依赖外部环境如果出错通常是代码逻辑错误的情况下不论我们如何重试都会继续错误的业务逻辑可以将该参数设置为0避免不必要的重试影响消息处理的速度。
深入思考
完成了上面的基础尝试之后再思考下面两个问题
问题一如果在重试过程中消息处理成功了还会有异常信息吗
答案是不会。因为重试过程是消息处理的一个整体如果某一次重试成功了会任务对所收到消息的消费成功了。
这个问题可以在上述例子中做一些小改动来验证比如
Slf4jComponentstatic 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?messagehello可以获得如下日志结果并没有异常打印出来。
2018-12-10 16:07:38.390 INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener : Received: hello, 12018-12-10 16:07:39.398 INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener : Received: hello, 22018-12-10 16:07:41.402 INFO 66468 --- [L6MGAj-MAj7QA-1] c.d.stream.TestApplication$TestListener : Received: hello, 3也就是虽然前两次消费抛出了异常但是并不影响最终的结果也不会打印中间过程的异常避免了对日志告警产生误报等问题。
问题二如果重试都失败之后应该怎么办呢
如果消息在重试了还是失败之后目前的配置唯一能做的就是将异常信息记录下来进行告警。由于日志中有消息的消息信息描述所以应用维护者可以根据这些信息来做一些补救措施。
当然这样的做法显然不是最好的因为太过麻烦。那么怎么做才好呢且听下回分解
代码示例
本文示例读者可以通过查看下面仓库的中的stream-exception-handler-1项目
GithubGitee
如果您对这些感兴趣欢迎star、follow、收藏、转发给予支持
以下专题教程也许您会有兴趣
Spring Boot基础教程Spring Cloud基础教程
