苏州网站开发外包公司,免费ppt模板下载公众号,模版型网站是怎样的,wordpress主机主题应用场景 上一篇《Spring Cloud Stream消费失败后的处理策略#xff08;一#xff09;#xff1a;自动重试》介绍了默认就会生效的消息重试功能。对于一些因环境原因、网络抖动等不稳定因素引发的问题可以起到比较好的作用。但是对于诸如代码本身存在的逻辑错误等#xff…应用场景 上一篇《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 : 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-input.groupstream-exception-handlerspring.cloud.stream.bindings.example-topic-input.consumer.max-attempts1spring.cloud.stream.bindings.example-topic-output.destinationtest-topic完成了上面配置之后启动应用并访问localhost:8080/sendMessage?messagehello接口来发送一个消息到MQ中了此时可以看到消费失败后抛出了异常跟上一篇文章的结果一样消息消费失败记录了日志消息信息丢弃。 下面针对消息消费失败在TestListener中针对消息消费逻辑创建一段错误处理逻辑比如 Slf4jComponentstatic class TestListener { StreamListener(TestTopic.INPUT) public void receive(String payload) { log.info(Received payload : payload); throw new RuntimeException(Message consumer failed!); } /** * 消息消费失败的降级处理逻辑 * * param message */ ServiceActivator(inputChannel test-topic.stream-exception-handler.errors) public void error(Message? message) { log.info(Message consumer failed, call fallback!); }}通过使用ServiceActivator(inputChannel test-topic.stream-exception-handler.errors)指定了某个通道的错误处理映射。其中inputChannel的配置中对应关系如下 test-topic消息通道对应的目标destination即spring.cloud.stream.bindings.example-topic-input.destination的配置stream-exception-handler消息通道对应的消费组group即spring.cloud.stream.bindings.example-topic-input.group的配置 再启动应用并访问localhost:8080/sendMessage?messagehello接口来发送一个消息到MQ中此时可以看到日志如下 2018-12-11 12:00:35.500 INFO 75269 --- [ctor-http-nio-3] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:5672]2018-12-11 12:00:35.512 INFO 75269 --- [ctor-http-nio-3] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory.publisher#311db1cb:0/SimpleConnection40370d8c [delegateamqp://guest127.0.0.1:5672/, localPort 54391]2018-12-11 12:00:35.527 INFO 75269 --- [ption-handler-1] c.d.stream.TestApplication$TestListener : Received: hello,2018-12-11 12:00:38.541 INFO 75269 --- [ption-handler-1] c.d.stream.TestApplication$TestListener : Message consumer failed, call fallback!虽然消费逻辑中输出了消息内容之后抛出了异常但是会进入到error函数中执行错误处理逻辑这里只是答应了一句话用户可以根据需要读取消息内容以及异常详情做更进一步的细化处理。 深入思考 由于error逻辑是通过编码方式来实现的所以这段逻辑相对来说比较死。通常只有业务上有明确的错误处理逻辑的时候这种方法才可以比较好的被应用到。不然能做的可能也只是将消息记录下来然后具体的分析原因后再去做补救措施。所以这种方法也不是万能的主要适用于有明确错误处理方案的方式来使用这种场景并不多另外。。。 注意有坑 这个方案在目前版本2.0.x其实还有一个坑这种方式并不能很好的处理异常消息会有部分消息得不到正确的处理由于应用场景也不多所以目前不推荐使用这种方法来做完全可以用原始的异常捕获机制来处理只是没有这种方式那么优雅。目前看官方issue是在Spring Cloud Stream的2.1.0版本中会修复后续发布之后可以使用该功能具体点击查看Issue #1357。 而对于没有特定的错误处理方案的也只能通过记录和后续处理来解决可能这样的方式也只是比从日志中抓去简单那么一些并没有得到很大的提升。但是不要紧因为下一篇我们将继续介绍其他更好的处理方案。 代码示例 本文示例读者可以通过查看下面仓库的中的stream-exception-handler-2项目 GithubGitee 如果您对这些感兴趣欢迎star、follow、收藏、转发给予支持 以下专题教程也许您会有兴趣 Spring Boot基础教程Spring Cloud基础教程