1 前言
2 概述
3 核心概念
- 3.1 绑定器
- 3.2 发布/订阅模型
4 Spring Cloud Stream标准流程套路
5 入门案例
6 自定义消息通道
- 6.1 概述
- 6.2 自定义消息通道
  - 6.2.1 消息生产者
  - 6.2.2 消息消费者
7 消息分组
8 消息分区

1 前言

在实际的企业开发中，消息中间件是至关重要的组件之一。消息中间件主要解决应用解耦、异步消息、流量削峰等问题，实现高性能、高可用、可伸缩和最终一致性的架构。不同的中间件其实现方式，内部结构是不一样的。如常见的RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，如RabbitMQ有Exchange，Kafka有Topic、Partitions，这些中间件的差异性导致我们实际项目开发会给我们造成了一定的困扰，我们如果用了两个消息中间件的其中一种，后面的业务需求，需要往另一种消息中间件迁移，这时候无疑就是一个灾难，一大堆东西需要重新推倒重做，因为它和我们系统耦合了，这时候Spring Cloud Stream给我们提供了一种解耦合的方式。

2 概述

Spring Cloud Stream由一个中间件中立的Core组成。应用通过Spring Cloud Stream的input（相当于消费者consumer，它是从队列中接收消息的）和output（相当于生产者producer，它是从队列中发送消息的）通道和外界交流。通道通过指定中间件的Binder实现和外部代理连接。业务开发者不再关注具体消息中间件，只需要关注Binder对应用程序提供的抽象概念来使用消息中间件实现业务即可。

Spring Cloud Stream处理架构.png

3 核心概念

3.1 绑定器

绑定器（Binder）是Spring Cloud Stream中的一个非常重要的概念。在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接和消息中间件进行信息交互的时候，由于各个消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性，这似的我们实现的消息交互逻辑就会非常笨重，因此对具体的中间件实现细节有太重的依赖，当中间件有较大的变动升级、或者更换中间件的时候，我们就需要付出非常大的代价来实施。
通过定义绑定器作为中间层，实现了应用程序和消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel，使得应用程序不需要再考虑各种不同的消息中间件实现。当需要升级消息中间件或者更换其他的消息中间件产品时，我们需要做的就是更换对应的Binder绑定器而不需要修改任何应用逻辑，甚至可以任意的改变中间件的类型而不需要修改一行代码。
Spring Cloud Stream支持各种Binder实现，如下图所示：

Spring Cloud Stream支持各种Binder实现.png

通过配置把应用和Spring Cloud Stream的binder绑定在一起，之后我们只需要修改binder的配置来达到动态修改Topic、Exchange、type等一系列信息而不需要修改一行代码。

3.2 发布/订阅模型

在Spring Cloud Stream中的消息通信方式遵循了发布—订阅模式，当一条消息被投递到消息中间件之后，它会通过共享的Topic主题进行广播，消息消费者在订阅的主题中收到它并触发自身的业务逻辑处理。这里所提到的Topic主题是Spring Cloud Stream中的一个抽象概念，用来代表发布共享消息给消费者的地方。在不通过的消息中间件中，Topic可能对应着不同的概念，如：在RabbitMQ中它对应了Exchange，在Kafka中对应了Topic。

4 Spring Cloud Stream标准流程套路

Spring Cloud Stream标准流程套路1.png

Middleware：中间件，目前只支持RabbitMQ和Kafka。
Binder：Binder是应用和消息中间件之间的封装，目前实现了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型（对应于Kafka的Topic和RabbitMQ的Exchange），这些都是可以通过配置文件来实现。
@Input：注解标识输入通道，通过该输入通道接收到的消息进入应用程序。
@Output：注解标识输出通道，发布的消息将通过该通道离开应用程序。
@StreamListener：监听队列，用于消费者的队列的消息接收。
@EnableBinding：将通道Channel和Exchange绑定在一起。

Spring Cloud Stream标准流程套路2.png

Binder：很方便的连接中间件，屏蔽差异。
Channel：通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过对Channel实现对队列进行配置。
Source和Sink：简单可以理解为参照对象就是Spring Cloud Stream本身，从Spring Cloud Stream发布消息就是输出，从Spring Cloud Stream接受消息就是输入。

5 入门案例

5.1 准备工作

采用RabbitMQ作为消息中间件，完成Spring Cloud Stream的案例，Docker安装方式如下：

docker run -d  --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq:management

5.2 消息生产者

5.2.1 导入相关的依赖

修改部分：

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

完整部分：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://maven.apache.org/POM/4.0.0"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>spring_cloud_demo</artifactId>
        <groupId>org.sunxiaping</groupId>
        <version>1.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>stream_producer9005</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
</project>

5.2.2 修改配置文件

application.yml

server:
  port: 9005
spring:
  application:
    name: stream-producer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        output:
          destination: rabbit-default # 指定消息发送的目的地。在RabbitMQ中，发送到一个名为rabbit-default的Exchange上。
          contentType: application/json # 消息的类型
          binder: defaultRabbit

5.2.3 配置启动类

启动类：

package com.sunxiaping.stream.producer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * 发送消息，需要定义一个通道接口，通过接口中内置的Message Channel。
 * Spring Cloud Stream中内置接口 Source
 * `@EnableBinding`注解：绑定对应通道
 * 发送消息的话，通过MessageChannel发送消息
 *
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 20:48
 */
@SpringBootApplication
public class StreamProducer9005Application {
    public static void main(String[] args) {
        SpringApplication.run(StreamProducer9005Application.class, args);
    }
}

5.2.4 编写业务逻辑

IMessageProvider.java

package com.sunxiaping.stream.producer.service;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:25
 */
public interface IMessageProvider {
    void send();
}

MessageProviderImpl.java

package com.sunxiaping.stream.producer.service.impl;
import com.sunxiaping.stream.producer.service.IMessageProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import java.util.UUID;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:25
 */
@EnableBinding(Source.class)
public class MessageProviderImpl implements IMessageProvider {
    @Autowired
    private MessageChannel output;
    @Override
    public void send() {
        //通过MessageBuilder帮助我们创建消息
        this.output.send(MessageBuilder.withPayload(UUID.randomUUID().toString()).build());
    }
}

SendMessageController.java

package com.sunxiaping.stream.producer.controller;
import com.sunxiaping.stream.producer.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:33
 */
@RestController
public class SendMessageController {
    @Resource
    private IMessageProvider messageProvider;
    @GetMapping(value = "/sendMessage")
    public String sendMessage() {
        messageProvider.send();
        return "发送成功";
    }
}

5.2.5 执行流程

消息生产者的执行流程.png

5.3 消息消费者

5.3.1 导入依赖

修改部分：

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

完整部分：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://maven.apache.org/POM/4.0.0"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>spring_cloud_demo</artifactId>
        <groupId>org.sunxiaping</groupId>
        <version>1.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>stream_consumer9006</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
</project>

5.3.2 修改配置文件

application.yml：

server:
  port: 9006
spring:
  application:
    name: stream-consumer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        input:
          destination: rabbit-default # 指定消息获取的目的地。
          contentType: application/json # 消息的类型
          binder: defaultRabbit

5.3.3 配置启动类

启动类：

package com.sunxiaping.stream.consumer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-09 11:09
 */
@SpringBootApplication
public class StreamConsumer9006Application {
    public static void main(String[] args) {
        SpringApplication.run(StreamConsumer9006Application.class, args);
    }
}

5.3.4 编写业务逻辑

MessageConsumer.java

package com.sunxiaping.stream.consumer.listener;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-09 11:12
 */
@EnableBinding(Sink.class)
public class MessageConsumer {
    //监听binding中的消息
    @StreamListener(Sink.INPUT)
    public void receive(Message<String> message) {
        System.out.println("message = " + message.getPayload());
    }
}

6 自定义消息通道

6.1 概述

Spring Cloud Stream内置类两种接口，分别定义了binding为“input”输入流和“output”输出流，而在我们实际使用中，往往需要定义各种输入输出流。使用方法也很简单。

6.2 自定义消息通道

6.2.1 消息生产者

修改配置文件：

server:
  port: 9005
spring:
  application:
    name: stream-producer
  #    rabbitmq:
  #      host: 192.168.32.100
  #      port: 5672
  #      username: guest
  #      password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        output:
          destination: rabbit-default # 指定消息发送的目的地。在RabbitMQ中，发送到一个名为rabbit-default的Exchange上。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
        # 自定义通道  
        customOutput:
          destination: rabbit-custom

CustomProcessor.java

package com.sunxiaping.stream.producer.channel;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
 * 自定义消息通道
 *
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-09 13:20
 */
public interface CustomProcessor extends Source, Sink {
    /**
     * 自定义输出通道
     */
    String CUSTOM_OUTPUT = "customOutput";
    /**
     * 自定义输入通道
     */
    String CUSTOM_INPUT = "customInput";
    /**
     * 消息生产者的配置
     *
     * @return
     */
    @Output(CustomProcessor.CUSTOM_OUTPUT)
    MessageChannel customOutput();
    /**
     * 消息消费者的配置
     *
     * @return
     */
    @Input(CustomProcessor.CUSTOM_INPUT)
    SubscribableChannel customInput();
}

IMessageProvider.java

package com.sunxiaping.stream.producer.service;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:25
 */
public interface IMessageProvider {
    void send();
}

MessageProviderImpl.java

package com.sunxiaping.stream.producer.service.impl;
import com.sunxiaping.stream.producer.channel.CustomProcessor;
import com.sunxiaping.stream.producer.service.IMessageProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import java.util.UUID;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:25
 */
@EnableBinding(CustomProcessor.class)
public class MessageProviderImpl implements IMessageProvider {
    @Autowired
    @Qualifier(value = "customOutput")
    private MessageChannel customOutput;
    @Override
    public void send() {
        //通过MessageBuilder帮助我们创建消息
        this.customOutput.send(MessageBuilder.withPayload(UUID.randomUUID().toString()).build());
    }
}

SendMessageController.java

package com.sunxiaping.stream.producer.controller;
import com.sunxiaping.stream.producer.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-08 22:33
 */
@RestController
public class SendMessageController {
    @Resource
    private IMessageProvider messageProvider;
    @GetMapping(value = "/sendMessage")
    public String sendMessage() {
        messageProvider.send();
        return "发送成功";
    }
}

6.2.2 消息消费者

修改配置文件：

server:
  port: 9006
spring:
  application:
    name: stream-consumer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        input:
          destination: rabbit-default # 指定消息获取的目的地。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
        # 自定义通道  
        customInput:
          destination: rabbit-custom

CustomProcessor.java

package com.sunxiaping.stream.consumer.channel;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

/**
 * 自定义通道
 *
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-09 13:41
 */
public interface CustomProcessor extends Source, Sink {

    /**
     * 自定义输出通道
     */
    String CUSTOM_OUTPUT = "customOutput";
    /**
     * 自定义输入通道
     */
    String CUSTOM_INPUT = "customInput";

    /**
     * 消息生产者的配置
     *
     * @return
     */
    @Output(CustomProcessor.CUSTOM_OUTPUT)
    MessageChannel customOutput();

    /**
     * 消息消费者的配置
     *
     * @return
     */
    @Input(CustomProcessor.CUSTOM_INPUT)
    SubscribableChannel customInput();
}

MessageConsumer.java

package com.sunxiaping.stream.consumer.listener;

import com.sunxiaping.stream.consumer.channel.CustomProcessor;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

/**
 * @author 许大仙
 * @version 1.0
 * @since 2020-10-09 11:12
 */
@EnableBinding(CustomProcessor.class)
public class MessageConsumer {

    //监听binding中的消息
    @StreamListener(CustomProcessor.CUSTOM_INPUT)
    public void receive(Message<String> message) {
        System.out.println("message = " + message.getPayload());
    }
}

7 消息分组

通常在生产环境中，我们的每个服务都不会以单节点的方式，当同一个服务启动多个实例的时候，这些实例都会绑定到同一个消息通道的目标主题（Topic）上。默认情况下，当生产者发出一条消息到绑定通道上，这条消息会产生多个副本被每个消费者实例接收和处理，但是有些业务场景下，我们希望生产者的消息只被一个实例消费，这个时候我们需要为这些消费者设置消费组来实现这样的功能。

消息分组.png

实现的方式很简单，我们只需要在服务消费端设置spring.cloud.stream.bindings.input.group属性即可，如下配置：

server:
  port: 9006

spring:
  application:
    name: stream-consumer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        input:
          destination: rabbit-default # 指定消息获取的目的地。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
          group: group1 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）
        # 自定义通道
        customInput:
          destination: rabbit-custom
          group: group2 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）

server:
  port: 9007

spring:
  application:
    name: stream-consumer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        input:
          destination: rabbit-default # 指定消息获取的目的地。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
          group: group1 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）
        # 自定义通道
        customInput:
          destination: rabbit-custom
          group: group2 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）

在同一个group中的多个消费者只有一个可以获取到消息并消费。

8 消息分区

有一些场景需要满足，同一特征的数据被同一个实例消费，比如同一个id的传感器监测数据必须被同一个实例统计计算分组，否则可能无法获取全部的数据。又比如部分异步任务，首次请求启动task，二次请求取消task，此场景就必须保证两次请求到同一个实例。

消息分区.png

消息生产者配置：

server:
  port: 9005

spring:
  application:
    name: stream-producer
  #    rabbitmq:
  #      host: 192.168.32.100
  #      port: 5672
  #      username: guest
  #      password: guest
  cloud:
    stream:
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        output:
          destination: rabbit-default # 指定消息发送的目的地。在RabbitMQ中，发送到一个名为rabbit-default的Exchange上。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
        # 自定义通道
        customOutput:
          destination: rabbit-custom
          producer:
            # ===========修改部分开始===========
            partition-key-expression: payload #通过该参数指定了分组键的表达式规则，我们可以根据实际的输出消息规则来配置SPEL来生成合适的分区键
            partition-count: 2 # 该参数指定了消息分区的数量
            # ===========修改部分结束===========

消息消费者配置：

server:
  port: 9006

spring:
  application:
    name: stream-consumer
  #  rabbitmq:
  #    host: 192.168.32.100
  #    port: 5672
  #    username: guest
  #    password: guest
  cloud:
    stream:
      # ===========修改部分开始===========
      instance-count: 2 # 该参数指定了当前消费者实例的数量
      instance-index: 0 # 该参数设置当前实例的索引号，从0开始，最大值为spring.stream.instance-count-1
      # ===========修改部分结束===========
      binders: # 配置绑定器
        defaultRabbit: # 表示定义的名称，用于binding整合
          type: rabbit # 消息组件类型
          environment: # 设置RabbitMQ的相关的环境配置
            spring:
              rabbitmq:
                host: 192.168.32.100
                port: 5672
                username: guest
                password: guest
      bindings: # 绑定通道
        input:
          destination: rabbit-default # 指定消息获取的目的地。
          contentType: application/json # 消息的类型
          binder: defaultRabbit
          group: group1 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）
        # 自定义通道
        customInput:
          destination: rabbit-custom
          group: group2 # 设置消息的组名（同名组中的多个消息者，只有一个去消费消息）
          # ===========修改部分开始===========
          consumer:
            partitioned: true # 开启消费者分区功能
          # ===========修改部分结束===========