官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。

应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互，通过我们配置来 binding ，而 Spring Cloud Stream 的 binder 负责与消息中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。
通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前仅支持RabbitMQ、Kafka。
这里还要讲解一下什么是Spring Integration ？ Integration 集成

企业应用集成（EAI）是集成应用之间数据和服务的一种应用技术。四种集成风格：

1.文件传输：两个系统生成文件，文件的有效负载就是由另一个系统处理的消息。该类风格的例子之一是针对文件轮询目录或FTP目录，并处理该文件。

2.共享数据库：两个系统查询同一个数据库以获取要传递的数据。一个例子是你部署了两个EAR应用，它们的实体类（JPA、Hibernate等）共用同一个表。

3.远程过程调用：两个系统都暴露另一个能调用的服务。该类例子有EJB服务，或SOAP和REST服务。

4.消息：两个系统连接到一个公用的消息系统，互相交换数据，并利用消息调用行为。该风格的例子就是众所周知的中心辐射式的（hub-and-spoke）JMS架构。

为什么需要SpringCloud Stream消息驱动呢？

比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，像RabbitMQ有exchange，kafka有Topic，partitions分区，这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了，这时候springcloud Stream给我们提供了一种解耦合的方式。

Spring Cloud Stream 是一个构建消息驱动微服务的框架。应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互，通过我们配置来 binding ，而 Spring Cloud Stream 的 binder 负责与中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。

Spring Cloud Stream由一个中间件中立的核组成。应用通过Spring Cloud Stream插入的input(相当于消费者consumer，它是从队列中接收消息的)和output(相当于生产者producer，它是从队列中发送消息的。)通道与外界交流。

通道通过指定中间件的Binder实现与外部代理连接。业务开发者不再关注具体消息中间件，只需关注Binder对应用程序提供的抽象概念来使用消息中间件实现业务即可。

springcloudStream.png

• Binder

Binder 是 Spring Cloud Stream 的一个抽象概念，是应用与消息中间件之间的粘合剂。目前 Spring Cloud Stream 实现了 Kafka 和 Rabbit MQ 的binder。

通过 binder ，可以很方便的连接中间件，可以动态的改变消息的destinations（对应于 Kafka 的topic，Rabbit MQ 的 exchanges），这些都可以通过外部配置项来做到。甚至可以任意的改变中间件的类型而不需要修改一行代码。

• Publish-Subscribe

消息的发布（Publish）和订阅（Subscribe）是事件驱动的经典模式。Spring Cloud Stream 的数据交互也是基于这个思想。生产者把消息通过某个 topic 广播出去（Spring Cloud Stream 中的 destinations）。其他的微服务，通过订阅特定 topic 来获取广播出来的消息来触发业务的进行。

这种模式，极大的降低了生产者与消费者之间的耦合。即使有新的应用的引入，也不需要破坏当前系统的整体结构。

• Consumer Groups

“Group”，如果使用过 Kafka 的童鞋并不会陌生。Spring Cloud Stream 的这个分组概念的意思基本和 Kafka 一致。

微服务中动态的缩放同一个应用的数量以此来达到更高的处理能力是非常必须的。对于这种情况，同一个事件防止被重复消费，只要把这些应用放置于同一个 “group” 中，就能够保证消息只会被其中一个应用消费一次。

• Consumer Groups

bindings 是我们通过配置把应用和spring cloud stream 的 binder 绑定在一起，之后我们只需要修改 binding 的配置来达到动态修改topic、exchange、type等一系列信息而不需要修改一行代码。

Demo演练

这里新建3个模块stream-Publish、stream-Subscribe1、stream-Subscribe2，其中stream-Publish作为消息发布模块，stream-Subscribe1和stream-Subscribe2作为消息消费模块（没有kafka的，windows下载链接：http://kafka.apache.org/downloads内带zookeeper）
springcloud版本：Hoxton.SR3

3个项目都需要以下依赖

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

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

stream-Publish的yml配置

server:
  port: 7888
spring:
  application:
    name: producer
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092         #Kafka的消息中间件服务器
          zk-nodes: localhost:2181        #Zookeeper的节点，如果集群，后面加,号分隔
          auto-create-topics: true        #如果设置为false,就不会自动创建Topic 有可能你Topic还没创建就直接调用了。
      bindings:
        output:      #这里用stream给我们提供的默认output，后面会讲到自定义output        
            destination: stream-demo    #消息发往的目的地            
            content-type: text/plain    #消息发送的格式，接收端不用指定格式，但是发送端要

stream-Publish的SendService

@EnableBinding(Source.class)
public class SendService {
     @Autowired
        private Source source;
     
     public void sendMsg(String msg) { h
         source.output().send(MessageBuilder.withPayload(msg).build());
     }
}

stream-Publish的测试Controller

@RestController
@SpringBootApplication
public class DemoApplication {
    
     @Autowired
        private SendService sendService;
     
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
    
    @GetMapping("/send/{msg}")
    public void send(@PathVariable("msg") String msg){
        sendService.sendMsg(msg);
    }

}

然后需要来写消息消费者的代码，stream-Subscribe1、stream-Subscribe2（同前一个，换个端口）
stream-Subscribe1的yml文件

server:
  port: 7889
spring:
  application:
    name: consumer_1
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          zk-nodes: localhost:2181
          auto-create-topics: true
      bindings:
#input是接收，注意这里不能再像前面一样写output了
          input:
            destination: stream-demo

stream-Subscribe1的接受消息的Service（启动类默认就OK）

@EnableBinding(Sink.class)
public class RecieveService {
 
    @StreamListener(Sink.INPUT)
    public void recieve(Object payload){
        System.out.println(payload);
    }
}

然后我们启动zookeeper，和Kafka，然后启动这三个项目
访问我们stream-Publish的测试接口，如下：

测试.png

请求成功后我们可以看到消息消费者的控制台打印出消息hello

image.png

自定义信道

好了到现在为止，我们进行了一个简单的消息发送和接收，用的是Stream给我们提供的默认Source，Sink，接下来我们要自己进行自定义，这种方式在工作中还是用的比较多的，因为我们要往不同的消息通道发消息，必然不能全都叫input,output的，那样的话就乱套了，因此首先自定义一个接口，如下：

• Source（发射器）: 一个接口类，内部定义了一个输出管道，例如定义一个输出管道 @output（"XXOO"）。说明这个发射器将会向这个管道发射数据。

• Sink（接收器）:一个接口类，内部定义了一个输入管道，例如定义一个输入管道 @input（"XXOO"）。说明这个接收器将会从这个管道接收数据。

• Binder（绑定器）:用于与管道进行绑定。Binder将于消息中间件进行关联。@ EnableBinding （Source.class/Sink.class）。@EnableBinding（）里面是可以定义多个发射器/接收器

自定义MySource：
在stream-Publish中创建自定义接口

public interface MySource {    
    @Output("myOutput")//管道名称为"myOutput,对应在yml文件里
    MessageChannel myOutput();
}

修改stream-Publish中的SendService文件

@EnableBinding(MySource.class) //使用我们自定义的Mysource
public class SendService {
     @Autowired
        private MySource mySource;
     
     public void sendMsg(String msg) { 
         mySource.myOutput().send(MessageBuilder.withPayload(msg).build());
     }
}

修改stream-Publish中yml文件

server:
  port: 7888
spring:
  application:
    name: producer
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092         #Kafka的消息中间件服务器
          zk-nodes: localhost:2181        #Zookeeper的节点，如果集群，后面加,号分隔
          auto-create-topics: true        #如果设置为false,就不会自动创建Topic 有可能你Topic还没创建就直接调用了。
      bindings:
        myOutput:   #自定义output
            destination: stream-demo    #消息发往的目的地
            content-type: text/plain    #消息发送的格式，接收端不用指定格式，但是发送端要      #这里用stream给我们提供的默认output，后面会讲到自定义output        

到这里，我们的消息发送服务已经修改完啦，接下来修改消息消费服务stream-Subscribe1，类似于上面，之类我直接贴代码了：

public interface MySink {
    @Input("myInput")
    SubscribableChannel myInput();
}

@EnableBinding(MySink.class)
public class RecieveService {
 
    @StreamListener("myInput")
    public void recieve(Object payload){
        System.out.println(payload);
    }
}

server:
  port: 7889
spring:
  application:
    name: consumer_1
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          zk-nodes: localhost:2181
          auto-create-topics: true
      bindings:
#修改为我们自己的myInput
          myInput:
            destination: stream-demo

然后测试成功，和上面一样就不放图了

消息分组(Consumer Groups)

“Group”，如果使用过 Kafka 的读者并不会陌生。Spring Cloud Stream 的这个分组概念的意思基本和 Kafka 一致。微服务中动态的缩放同一个应用的数量以此来达到更高的处理能力是非常必须的。对于这种情况，同一个事件防止被重复消费，

只要把这些应用放置于同一个 “group” 中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。

我们只需要修改yml文件就可以啦，两个消费者服务都配置相同的名称的group

server:
  port: 7889
spring:
  application:
    name: consumer_1
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          zk-nodes: localhost:2181
          auto-create-topics: true
      bindings:
          input:
            destination: stream-demo
            group: group #加上一条group就可以啦

可以看到stream-Subscribe1和stream-Subscribe2是属于同一组的。springcloud-stream模块的发的消息只能被stream-Subscribe1或stream-Subscribe2其中一个接收到，这样避免了重复消费。

消息分区

Spring Cloud Stream对给定应用的多个实例之间分隔数据予以支持。在分隔方案中，物理交流媒介（如：代理主题）被视为分隔成了多个片（partitions）。一个或者多个生产者应用实例给多个消费者应用实例发送消息并确保相同特征的数据被同一消费者实例处理。

Spring Cloud Stream对分割的进程实例实现进行了抽象。使得Spring Cloud Stream 为不具备分区功能的消息中间件（RabbitMQ）也增加了分区功能扩展。

那么我们就要进行一些配置了，比如我只想要stream-Subscribe2模块接收到消息，stream-Subscribe2配置如下：

server:
  port: 7890
spring:
  application:
    name: consumer_2
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          zk-nodes: localhost:2181
          auto-create-topics: true
      bindings:
          input:
            destination: stream-demo
            group: group
            consumer:
              partitioned: true   #开启分区
      instance-count: 2     #分区数量

stream-Publish模块配置如下：

server:
  port: 7888
spring:
  application:
    name: producer
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          zk-nodes: localhost:2181
          auto-create-topics: true
      bindings:
        myOutput:
            destination: stream-demo
            content-type: text/plain
            producer:
              partitionKeyExpression:  payload.id(你自己的key)   #分区的主键，根据什么来分区，下面的payload.id只是一个对象的id用于做为Key，用来说明的。希望不要误解
              partitionCount: 2    #Key和分区数量进行取模去分配消息，这里分区数量配置为2

其他的代码基本不变，这里就不演示了。这里要给大家说明一下，比如分区的Key是一个对象的id，比如说id=1,每次发送消息的对象的id为相同值1，则消息只会被同一个消费者消费，比如说Key和分区数量取模计算的结果是分到stream2模块中，那么下一次进行进行消息发送，

只要分组的key即id的值依然还是1的话，消息永远只会分配到stream2模块中。

参考：https://blog.csdn.net/JinXYan/article/details/90813592