1. 概述

Spring Cloud Stream是一个构建消息驱动的微服务框架，屏蔽底层消息中间件的差异,降低切换成本，统一消息的编程模型。应用程序通过inputs或者outputs来与Spring Cloud Stream中binder对象交互，通过配置的binding绑定，其中Spring Cloud Stream的binder对象负责与消息中间件交互，通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念，目前仅支持RabbitMQ、Kafka
实现原理：通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离
Binder可以生成Binding，Binding用来绑定消息容器的生产者和消费者，它有两种类型，INPUT和OUTPUT，INPUT对应于消费者，OUTPUT对应于生产者

Binder：连接中间件，屏蔽差异
Channel：是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置
Source和Sink：从Stream发布消息就是输出，接受消息就是输入
@Input：输入通道，通过该输入通道接收到的消息进入应用程序
@Output：输出通道，发布的消息将通过该通道离开应用程序
@StreamListener：监听队列，用于消费者的队列的消息接收
@EnableBinding：信道channel和exchange绑定在一起
本文以RabbitMQ为例，需要提前搭建好RabbitMQ开发环境

2. 搭建生产者

2.1. 引入核心依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

2.2. 配置application.yml文件

server:
  port: 8830
spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
      bindings:
        output:
          destination: streamExchange
          content-type: application/json
          binder: defaultRabbit
  rabbitmq:
    host: xx.xx.xxx.xxx
    port: 5672
    username: admin
    password: 123456
eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka
  instance:
    lease-renewal-interval-in-seconds: 30 #设置心跳间隔
    lease-expiration-duration-in-seconds: 90 #间隔超时
    instance-id: ${spring.cloud.client.ip-address}:${server.port}
    prefer-ip-address: true

2.3. 编写主启动类

@SpringBootApplication
public class StreamProviderApplication {

    public static void main(String[] args) {
        SpringApplication.run(StreamProviderApplication.class, args);
    }
}

2.4. 编写业务接口类

@EnableBinding(Source.class)
public class MessageServiceImpl implements MessageService {

    @Resource
    private MessageChannel output;

    private static final Logger logger = LoggerFactory.getLogger(MessageServiceImpl.class);

    @Override
    public String sendMessage() {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());
        logger.info("生产者发送消息成功!");
        return serial;
    }
}

2.5. 验证

启动Eureka Server和stream provider微服务，浏览器地址栏输入地址http://localhost:8830/sendMessage

3. 搭建消费者

3.1. 引入核心依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

3.2. 编写application.yml文件

server:
  port: 8831
spring:
  application:
    name: cloud-stream-consumer
  rabbitmq:
    host: xx.xx.xxx.xxx
    port: 5672
    username: admin
    password: 123456
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
      bindings:
        input:
          destination: streamExchange
          content-type: application/json
          binder: defaultRabbit
eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka
  instance:
    lease-renewal-interval-in-seconds: 30 #设置心跳间隔
    lease-expiration-duration-in-seconds: 90 #间隔超时
    instance-id: ${spring.cloud.client.ip-address}:${server.port}
    prefer-ip-address: true

3.3. 编写主启动类

@SpringBootApplication
public class StreamConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(StreamConsumerApplication.class, args);
    }
}

3.4. 编写业务监听类

@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(ReceiveMessageListener.class);

    @StreamListener(Sink.INPUT)
    public void input(Message<String> message) {
        logger.info("消费者接收到生产者发送的消息为: {}", message.getPayload());
    }
}

3.5. 验证

启动stream consumer微服务

4. 消息分组

如何有多个消费者时，会出现重复消费的问题，可以使用Stream中的消息分组来解决，同一组内会发生竞争关系，其中只有一个可以消费
在配置文件中加入分组配置

spring:
  cloud:
    stream:
      bindings:
        input:
          group: xlhj