1、简介

Spring Cloud Stream是一个用来为微服务应用构建 消息驱动 能力的框架。通过使用 Spring Cloud Strea m ，可以有效简化开发人员对消息中间件的使用复杂度，降低代码与消息中间件间的耦合度，屏蔽消息中间件 之 间的差异性，让开发人员可以有更多的精力关注于核心业务逻辑的处理。

主要有以下几个组件：

1）、目的地绑定器（Destination Binders）：负责提供与外部消息系统集成的组件。

2）、固定器（Bindings）：介于外部消息系统与应用程序间的桥梁 ，这个应用程序提供了生产者和消费者的消息 （由 Destination Binders 创建）。

3）、输入管道（Input Bindings）：消费者通过Input Bindings 连接 Binder ，而 Binder 与 MQ 连接，即消费者通过 Input Bindings 从 MQ 读取数据。

4）、输出管道（Output Bindings）：生产者通过Output Bindings 连接 Binder ，而 Binder 与 MQ 连接，即生产者通过 Output Bindings 向 MQ 写入数据。

5）、消息（Message）：生产者和消费者使用的规范数据结构，用于与 Binders 通信（从而通过外部消息系统与其他应用程序通信）。

2、具体应用示例1（MQ使用kafka）

引入依赖

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

2.1、生产者

配置文件

server:port: 8090
spring:cloud:stream:kafka:binder:brokers: 192.168.30.88:9092,192.168.30.89:9092bindings:producer-out-0:destination: topic1content-type: application/json

代码实现

@Autowired
private StreamBridge streamBridge;@GetMapping("/test/send")
public String sendMsg(@RequestParam("msg") String msg){Map<String , Object> map = new HashMap<>();map.put("tag", "tags");MessageHeaders headers = new MessageHeaders(map);// 封装消息Message<String> message = MessageBuilder.createMessage(msg, headers);//发送消息streamBridge.send("producer-out-0", message);return msg;
}

2.2、消费者

配置文件

server:port: 8091
spring:cloud:stream:kafka:binder:brokers: 192.168.30.88:9092,192.168.30.89:9092function:definition: consumer # 这个名称要和下面bindings的consumer-in-0第一个单词一样bindings:consumer-in-0:destination: topic1content-type: application/json

代码实现

// 向容器中添加Consumer<Message<String>>类型的bean即可
@Bean
public Consumer<Message<String>> consumer(){return msg -> {System.out.println("接收到消息：" + msg.getPayload());};
}

3、具体应用示例2（MQ使用Rocketmq）

引入依赖

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

代码实现

@Autowired
private StreamBridge streamBridge;@GetMapping("/test/send")
public String sendMsg(@RequestParam("msg") String msg){Map<String , Object> map = new HashMap<>();map.put(MessageConst.PROPERTY_TAGS, "tags");MessageHeaders headers = new MessageHeaders(map);// 封装消息Message<String> message = MessageBuilder.createMessage(msg, headers);//发送消息streamBridge.send("producer-out-0", message);return JSON.toJSONString(message);
}

3.2、消费者

配置文件：

server:port: 8091
spring:cloud:stream:rocketmq:binder:name-server: 192.168.30.88:9876function:definition: consumer # 这个名称要和下面bindings的consumer-in-0第一个单词一样bindings:consumer-in-0:destination: topic1content-type: application/json

代码实现:

// 向容器中添加Consumer<Message<String>>类型的bean即可
@Bean
public Consumer<Message<String>> consumer(){return msg -> {System.out.println("接收到消息：" + msg.getPayload());};
}

注：

1、在spring-cloud-stream 3.1.0之前的版本，还有采用定义Source、Sink等方式编写消息生产者和消费者，在3.1.0以后的版本中弃用@StreamListener的方式，而采用函数式编程的方式接入，使用StreamBrige来进行发送。

2、注意binding的名称命名规则

例如：上面的代码中定义的consumer。

# 输入:    <方法名> + -in- + <index>
# 输出:    <方法名> + -out- + <index>

总结：本文介绍Stream统一消息中间件的模型，给出基于kafka和Rocketmq两种消息中间件模型下的使用案例，以及给出废弃使用老版本的Source、Sink模式解释。帮助大家快速上手Stream的使用。

       本人是一个从小白自学计算机技术，对运维、后端、各种中间件技术、大数据等有一定的学习心得，想获取自学总结资料（pdf版本）或者希望共同学习，关注微信公众号：上了年纪的小男孩。后台回复相应技术名称/技术点即可获得。（本人学习宗旨：学会了就要免费分享）