Spring Cloud Stream之rocketmq

Spring Cloud Stream 是一个构建消息驱动微服务的框架。以下是一个简单的例子，展示如何使用 Spring Cloud Stream 与 RocketMQ 集成发送和接收消息。

首先，在你的 pom.xml 中添加依赖：

<dependencies>
    <!-- Spring Cloud Stream -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    </dependency>
</dependencies>
 
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

配置文件 application.yml：

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          namesrv-addr: localhost:9876 # RocketMQ NameServer 地址
      bindings:
        output:
          destination: test-topic # 指定消息发送的 Topic
        input:
          destination: test-topic # 指定监听的 Topic
          group: test-group # 设置消费组

生产者代码示例：

@EnableBinding(Source.class)
public class Producer {
    @Autowired
    private MessageChannel output;
 
    public void send(String content) {
        output.send(MessageBuilder.withPayload(content).build());
    }
}

消费者代码示例：

@EnableBinding(Sink.class)
public class Consumer {
    @StreamListener(Sink.INPUT)
    public void receive(String payload) {
        System.out.println("Received: " + payload);
    }
}

在这个例子中，我们定义了一个名为 test-topic 的 RocketMQ 主题，并在 Producer 类中通过 send 方法向该主题发送消息，在 Consumer 类中通过 @StreamListener 注解监听并接收消息。