【RocketMQ系列九】SpringCloudStream整合RocketMQ-阿里云开发者社区

【RocketMQ系列九】SpringCloudStream整合RocketMQ

2024-07-03 9

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 【RocketMQ系列九】SpringCloudStream整合RocketMQ

1. Spring Cloud Stream是什么？

Spring Cloud Stream是一个框架，用于构建与共享消息系统连接的高度可扩展的事件驱动微服务。

官网：https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/

官网概述：https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-preface-notable-deprecations

该框架提供了一个灵活的编程模型，该模型基于已经建立和熟悉的Spring习惯用法和最佳实践，包括对持久pub/sub 语义、消费者组和有状态分区的支持。

简单的理解就是Spring Cloud Stream 通过在上层定义统一消息的编程模型，屏蔽了底层消息中间件的差异，降低了使用成本。下图展示了Spring Cloud Stream的处理架构

Spring Cloud Stream的核心构建块（编程模型）是：

Destination Binders: 负责提供与外部消息传递系统集成的组件。Binders 可以生成Bindings。
**Bindings: ** 外部消息系统和应用程序之间的桥梁，提供消息的生产者和消费者（由目标绑定器创建）。即用来绑定消息生产者和消息消费者。它有两种类型，INPUT和OUTPUT，INPUT对应消费者，OUTPUT对应生产者。
Message: 生产者和消费者用于与目标绑定器（以及通过外部消息系统与其他应用程序）通信的规范的数据结构。

2. Spring Cloud Stream的执行流程

3. 注解代码实现

首先创建一个生产者项目 my-springcloud-rocketmq-producer 和一个消费者项目 my-springcloud-rocketmq-consumer。

本demo使用的版本号是 cloud 2021.0.5.0 +springboot 2.6.13

在 my-springcloud-rocketmq-producer 上的操作

3.1. 引入依赖

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

3.2 . 属性文件配置

spring:
  cloud:
    stream:
      bindings:
        output:
          destination: my-springcloud-stream-topic
      rocketmq:
        binder:
          name-server: 172.31.184.89:9876

3.3. 定义生产者

在MySpringcloudRocketmqProducerApplication 添加 @EnableBinding(Source.class) 注解。然后创建生产者。

@Component
public class MyProducer {
  @Resource
  private Source source;
  
  public void sendMessage(String msg) {
    // 封装消息头
    Map<String, Object> headers = new HashMap<>();
    headers.put(MessageConst.PROPERTY_TAGS, "tagA");
    // 创建消息对象
    Message<String> message = MessageBuilder.createMessage(msg, new MessageHeaders(headers));
    // 发送消息
    source.output().send(message);
  }
}

在 my-springcloud-rocketmq-consumer上的操作

3.4. 引入依赖同生产者

3.5. 配置文件修改

spring.cloud.stream.rocketmq.binder.name-server=172.31.184.89:9876
spring.cloud.stream.bindings.input.destination=my-springcloud-stream-topic
spring.cloud.stream.bindings.input.group=my-springcloud-stream-consume-group

3.6. 定义消费者

在MySpringcloudRocketmqConsumerApplication 类上添加 @EnableBinding(Sink.class)注解。

@Component
public class MyConsumer {
   @StreamListener(Sink.INPUT)
   public void processMessage(String message) {
      System.out.println("收到的消息=" + message);
   }
}