Spring Cloud Alibaba 实战(九) - Spring Cloud Stream

1 定义

• 一个用于构建消息驱动的微服务的框架

用人话说就是 : 致力于简化MQ通讯的框架

• 官方项目地址

2 编程模型

◆ Destination Binder (目标绑定器)

• 与消息中间件通讯的组件

◆ Destination Bindings (目标绑定)
Binding是链接应用程序跟消息中间件的桥梁,用于消息的消费和生产，由binder建立
◆ Message(消息)
可见该编程模型异常强大,短短几行代码,就实现了消息的对接和处理
input/output就是微服务接收和发出消息

下面开始对内容中心编码

3 编写生产者

• 添加依赖

• 在启动类添加注解

• 写配置

4 编写消费者

编码用户中心

• 添加依赖

• 启动类上添加注解

• 写配置

5 自定义接口

5.1 发送消息

• 新建mysource接口

• 启动类注解

• 写配置,注意要和接口中的名字一致

• 测试代码

注意,因为mybatis会扫描启动类注解上scan注解所限制路径下的全部接口,因此一旦有接口未被xml mapper,即抛异常,因此编码时必须将扫描注解范围限定死在mapper包下!

5.2 消费消息

用户中心编码

• 写接口

• 添注解

• 加配置

透过现象看本质

当咱们定义好Source/Sink接口后,在启动类使用EnableBinding指定了接口后,就会使用IOC建立对应名字的代理类,因此配置文件中也必须同名

消息过滤

• 推荐阅读

Spring Cloud Stream实现消息过滤消费

监控

记得多看端点哦!
output/input其实就是一个channel

排错依据的重要端点

• /actuator/bindings
• /actuator/channels
• /actuator/health

异常处理

• 推荐阅读

Spring Cloud Stream错误处理详解

整合RocketMQ实现分布式事务

Stream自己并未考虑分布式事务问题,都是RocketMQ的能力

重构生产者

对内容中心一顿操做:删除没必要要代码

• 自定义的MySource接口,由于Spring内置的就已经知足咱们的需求了

• 接着别忘了删除启动类中对他的引用
• 删除TestController中对应测试代码
• 清理yml中Spring消息编程模型整合RocketMQ的部分

• myoutput删除

• 修正以下

代码重构

改造ShareService

• 即改造如下代码(直接删除)

• 添加Source

• 开始使用source发送消息,可是send只能直接发送消息(或者带有超时)

而咱们以前使用rocketmqtemplate传递参数时能够带个arg

那如今咱们该怎么传arg呢???
记得前面埋下的伏笔,header也是颇有用处的!
咱们能够将要传的参数放入header中,以下:

rocketmqtemplate功成身退,咱们可使用stream编程模型彻底替代了

改造AddBonusTransactionListener

• 如今这里的arg是null了

• 须要从header中获取arg了(有坑,后面再说),在这里打个断点

• 完善配置(IDEA没法识别,但确实会生效),实现事务功能

• 注意上面的group名称要与下一致

• 启动内容中心
• 发送请求

• 发现dto实际上是字符串,并非DTO对象

• 因此继续修正代码

• 这样就是正常的对象了

由于从header中获取的都是字符串哦!切记!

重构消费者

对用户中心删除没必要要代码,与内容中心相似,再也不详述

• 删除

• 将MyTestStreamConsumer改成AddBonusStreamConsumer

重构以下

总结

• 推荐阅读

Spring Cloud Stream知识点盘点

参考

• Spring Cloud Stream官方文档
• 面向将来微服务:Spring Cloud Alibaba从入门到进阶

本文由博客一文多发平台 OpenWrite 发布！