Spring Cloud Alibaba 实战(九) - Spr
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
重构如下
