spring cloud stream 3.1.2 源码搭配rocketmq学习 (一)

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

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

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

先粗略梳理一下流程:

初始化

1. functionBindingRegistrar的Bean初始化, 根据applicaion.yaml中的配置把Function配置注册成Bean, 并且根据配置创建出对应的Input/Ouput的MessageChannel.

   (MessageChannel的具体实现方法会new MessageDispatcher的实现, MessageDispatcher是MessageChannel中的一个属性).

2. functionInitializer的Bean初始化.找到步骤1中注册的Bean, 并把对应的Function进行处理后注册进MessageDispatcher中
3. BindingLifecycle的Bean初始化并调用其中的start方法, 将MessageChannel传入, MessageChannel和rocketmq进行绑定.

调用

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

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

总结

1. 把Function配置进行处理生成对应的MessageChannel并把Fucntion注册进MessageDispatcher中.
2. 把MessageChannel和rocketmq绑定.
3. rocketmq接受到消息后发送到对应的MessageChannel中消费.

后续文章会从源码的角度一步步看上续流程

Wish.

Do.