SpringCloudStream+RocketMQ多topic

之前写过两篇关于SpringCloudStream文章

spring-cloud-stream版本升级，告别旧注解@EnableBinding，拥抱函数式编程_spring-cloud-stream output注解没有了-CSDN博客

SpringCloudStream+RocketMQ事务消息配置_spring-cloud-starter-stream-rocketmq-CSDN博客

提到了一个多topic问题，在不配置spring.cloud.stream.function.definition的情况下只能配置一个in和一个out，这篇文章来解决这个问题。

根据上面的第一篇文章续写代码

一、续写生产者

1.写配置

在bindings后面添加即可，这里给出完整配置

spring:application:name: providercloud:nacos:discovery:server-addr: localhost:8848namespace: localusername: nacospassword: nacosstream:rocketmq:binder:name-server: 127.0.0.1:9876bindings:## 新版本固定格式  channel名字-{out/in}-{index}addBounsChannel-out-0:destination: add-bounsgroup: bouns-producer-groupsecondChannel-out-0:destination: secondgroup: second-producer-groupsource-out-0:destination: thirdgroup: third-producer-group
server:port: 8081

2.写代码

同样的使用streamBridge.send发消息

public void sendMqSecondChannel() {streamBridge.send("secondChannel-out-0", "sendMqSecondChannel from Provider!");}

二、续写消费者

1.写配置

多个消费者时，需要在spring.cloud.stream.function.definition 定义bean名称，多个时用分号分隔

给出完整yaml代码

spring:application:name: consumercloud:nacos:discovery:server-addr: localhost:8848namespace: localusername: nacospassword: nacosstream:function:definition: addBounsChannel;secondChannel;sourcerocketmq:binder:name-server: 127.0.0.1:9876bindings:## 新版本固定格式  channel名字-{out/in}-{index}addBounsChannel-in-0:destination: add-bounsgroup: bouns-consumer-groupsecondChannel-in-0:destination: secondgroup: second-consumer-groupsource-in-0:destination: thirdgroup: third-consumer-group
server:port: 8082

2.写代码

跟之前一样，定义bean的Consumer

   @Beanpublic Consumer<String> secondChannel() {return message -> {System.out.println("secondChannel消费消息:" + message);};}

三、常见问题

1.有Consumer必定会有Supplier，为什么不用Supplier发消息

Supplier也需要配置成@Bean,Supplier也跟Consumer一样用的监听模式，启动后不做处理会一直发，不如streamBridge.send发送灵活，这篇文章也有Supplier发送的例子可供参考Kafka Streams with Spring Cloud Stream - Spring Cloud

2.本文完整项目示例代码

【免费】springcloud-stream-rocketmq多topic示例代码资源-CSDN文库