spring cloud stream 3.1.2 源码搭配rocketmq绑定学习

Posted 2021-05-18 MY1024-

新版建议用响应式函数编程 即Function/Supplier/Consumer方式, 后续简称为Function

文章中使用的外部消息中间件是rockermq, 所以后续外部消息中间件都称为rockermq.
如: (先从图简单看看stream和外部消息中间件之间的绑定关系)
-> (先从图简单看看stream和rocketmq之间的绑定关系)

先从图简单看看stream和rocketmq之间的绑定关系

先粗略梳理一下流程:

初始化

1. 根据配置文件或者扫描包把Function属性注册成Bean, 完成之后会创建对应的Input/Ouput的MessageChannel.
2. Function进行初始化, 找出上一步中注册的Bean中的functionDefinition进行相应的处理后注册进MessageDispatcher的Handlers中.
3. BindingLifecycle开始注册并调用start, 将MessageChannel和rocketmq进行绑定.

调用

1. rocketmq接收到消息.找到初始化第三步绑定的MessageChannel并把消息发送.
2. MessageChannel会调用子类中的方法找到对应的MessageDispatcher,调用初始化第二步中注册进Handlers的方法, 完成消息的消费.

以上就是一个简单版本的 spring cloud stream 和 rocketmq 的关联关系

Wish.

Do.