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创建对应名字的***类,所以配置文件中也必须同名
消息过滤
监控
记得多看端点哦!
output/input其实就是一个channel
排错依据的重要端点
- /actuator/bindings
- /actuator/channels
- /actuator/health
异常处理
整合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重构如下
