先从图简单看看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中消费.