文档章节

Spring Cloud Stream消费失败后的处理策略(二):自定义错误处理逻辑

程序猿DD
 程序猿DD
发布于 2018/12/15 19:43
字数 1430
阅读 46
收藏 0

应用场景

上一篇《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 : " + 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-input.group=stream-exception-handler
spring.cloud.stream.bindings.example-topic-input.consumer.max-attempts=1

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

完成了上面配置之后,启动应用并访问localhost:8080/sendMessage?message=hello接口来发送一个消息到MQ中了,此时可以看到消费失败后抛出了异常,跟上一篇文章的结果一样,消息消费失败,记录了日志,消息信息丢弃。

下面,针对消息消费失败,在TestListener中针对消息消费逻辑创建一段错误处理逻辑,比如:

@Slf4j
@Component
static 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?message=hello接口来发送一个消息到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/SimpleConnection@40370d8c [delegate=amqp://guest@127.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项目:

如果您对这些感兴趣,欢迎star、follow、收藏、转发给予支持!

以下专题教程也许您会有兴趣

本文首发:http://blog.didispace.com/spring-cloud-starter-finchley-7-3/

© 著作权归作者所有

共有 人打赏支持
程序猿DD
粉丝 415
博文 79
码字总数 104947
作品 4
闵行
私信 提问
Spring Cloud Stream应用与自定义RocketMQ Binder:编程模型

前言: 本文作者张天,节选自笔者与其合著的《Spring Cloud微服务架构进阶》,即将在八月出版问世。本文将其中Spring Cloud Stream应用与自定义Rocketmq Binder的内容抽取出来,主要介绍Spr...

aoho
2018/06/23
0
0
Spring Cloud (十五)Stream 入门、主要概念与自定义消息发送与接收

前言 不写随笔的日子仿佛就是什么都没有产出一般……上节说到要学Spring Cloud Bus,这里发现按照官方文档的顺序反而会更好些,因为不必去后边的章节去为当前章节去打基础,所以我们先学习S...

东北小狐狸
2018/07/31
0
0
Spring Cloud Stream如何消费自己生产的消息?

在上一篇《Spring Cloud Stream如何处理消息重复消费》中,我们通过消费组的配置解决了多实例部署情况下消息重复消费这一入门时的常见问题。本文将继续说说在另外一个被经常问到的问题:如果...

程序猿DD
2018/11/28
0
0
[Spring Cloud] 7 Spring Cloud Sleuth

Spring Cloud Sleuth 分布式链路跟踪 Spring Cloud Sleuth是Spring Cloud的分布式链路跟踪解决方案。 7.1 Terminology Spring Cloud Sleuth借鉴了Dapper的术语。 Span :最基本的工作单元。例...

秋雨霏霏
2017/02/08
0
2
Spring Cloud学习:03断路器(Hystrix)

1 Hystrix介绍 Spring Cloud Hystrix是分布式系统处理超时和错误的机制,如下图所示,分布式系统中某个用户请求依赖A、H、I、P服务。 当此请求并发超过50的时候,服务I处理速度变慢,但是服务...

寒武没有纪
2017/10/22
0
0

没有更多内容

加载失败,请刷新页面

加载更多

VMware下centos7.x 用yum快速搭建LAMP平台

实验环境: [root@nmserver-7 html]# cat /etc/redhat-release CentOS release 7.3.1611 (AltArch) [root@nmserver-7 html]# uname -aLinux nmserver-7.test.com 3.10.0-514.el7.cent......

皇冠小丑
11分钟前
0
0
搜索引擎(Solr-索引详解)

时间字段类型特别说明 Solr中提供的时间字段类型( DatePointField, DateRangeField,废除的TrieDateField )是以时间毫秒数来存储时间的。 要求字段值以ISO-8601标准格式来表示时间:YYYY-MM...

这很耳东先生
46分钟前
4
0
Java成神之路

1、基础篇 01、面向对象 → 什么是面向对象 面向对象、面向过程 面向对象的三大基本特征和五大基本原则 → 平台无关性 Java 如何实现的平台无关 JVM 还支持哪些语言(Kotlin、Groovy、JRuby...

asdf08442a
今天
3
0
dubbo源码分析-服务导出

简介 dubbo框架spring Schema扩展机制与Spring集成,在spring初始化时候加载dubbo的配置类。 dubbo服务导出的入口类是ServiceBean的onApplicationEvent方法 ServiceBean的继承关系如下 publ...

王桥修道院副院长
今天
2
0
QQ音乐的动效歌词是如何实践的?

本文由云+社区发表 作者:QQ音乐技术团队 一、 背景 1. 现状 歌词浏览已经成为音乐app的标配,展示和动画效果也基本上大同小异,主要是单行的逐字染色的卡拉OK效果和多行的滚动效果。当然,我...

腾讯云加社区
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部