SpringCloud项目中配置RocketMQ的实现步骤

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>

spring:
  cloud:
    stream:
      function:
        definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法，多个分号隔开）
      bindings:
        producer1-out-0:
          destination: producer_topic # topic消息主题
          content-type: application/json # 内容格式
        consumer1-in-0:
          destination: consumer_topic # topic消息主题
          content-type: application/json # 内容格式
          group: consumer-group # 消费者组
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876 # rocketmq服务地址
          vipChannelEnabled: true # 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用。）

public interface CustomChannelBinder {
    @Output("Topic-send-Output")
    MessageChannel sendChannel();

    @Input("Topic-TAG1-Input")
    MessageChannel testInputChannel1();

    @Input("Topic-TAG2-Input")
    MessageChannel testInputChannel2();
}

@EnableBinding({CustomChannelBinder.class})

@Autowired
private CustomChannelBinder channelBinder;

public void sendMessage(String message) {
    channelBinder.sendChannel().send(MessageBuilder.withPayload(message).build());
}

@Service
public class ReceiveService {
    @StreamListener(value = CustomChannelBinder.INPUT1)
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}