官网：https://cloud.spring.io/spring-cloud-stream/
API: https://docs.spring.io/spring-cloud-stream/docs/Fishtown.M1/api/
DOC：https://docs.spring.io/spring-cloud-stream/docs/Fishtown.M1/reference/htmlsingle/

读者话

通过近段时间的摸索, 发现网上的资料, 要么就照搬其他人的博客, 要么就翻译官方的DOC, 导致对新手来讲, 实用性不强.本人在学习stream的时候也踩到不少坑, 为了帮助后来者快速入门, 特写本文.

Spring Cloud Stream介绍

Spring Cloud Stream 是一个构建消息驱动微服务的框架，应用程序通过input通道或者output通道来与Spring Cloud Stream中binder交互，通过配置来binding. 而Spring Cloud Stream的binder负责与中间件交互, 消息的中间件有(RabbitMQ, Kafka, ActiveMQ).
本文以RabbitMQ作为中间件作为讲解(启动rabbitMQ: rabbitmq-plugins enable rabbitmq_management)

binder.jpg

为什么要需要Stream呢? Stream的使用场景

a: 异步(消息的发送方, 只需要发送一条消息出去, 就可以不管了, 至于怎处理, 则交给消息的订阅者去处理)
b: 流量削峰(**宝, **东大的电商平台, 需要stream去做削峰)
d: 应用解耦(发布者和订阅者相互间解耦)
d: 日志处理

消息中重要的概念 binder 绑定器

1: 发布, 订阅模式( 该模式特点是广播 )

应用间通信遵照发布-订阅模型, 消息通过共享主题进行广播.
下图显示了交互的Spring Cloud Stream 应用的典型布局未处理的传感数据发布到raw-sensor-data的Topic进行广播, Averages 和IngestHDFS同时订阅了此消息, 收到消息后触发自身的处理逻辑.
Topic可能对应不同的概念, 在RabbitMQ表示的是Exchange, Kafka中对应Topic.

发布订阅.png

2: 消费组

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

组.png

3: 消息分区

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

消息分区.png

本文的项目结构

为了方便各位客官理解, 我讲本项目分为producer(生产者), consumer(消费者), 项目结构图如下:

项目结构图.jpg

程序讲解

Producer

1. 引入依赖 (pom.xml)

本项目使用的是springcloud2.0.4,所以要引入该包, 为了从仓库中取数据快, 则使用了阿里的镜像.

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.4.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>
. .... ...
<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <spring-cloud.version>Finchley.SR1</spring-cloud.version>
</properties>
. .... ...
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

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

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.1.15</version>
</dependency>

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

<!--使用aliyun镜像-->
<repositories>
    <repository>
        <id>alimaven</id>
        <name>Maven Aliyun Mirror</name>
        <url>http://maven.aliyun.com/nexus/content/repositories/central/</url>
        <releases>
            <enabled>true</enabled>
        </releases>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

2: application.yml

程序的端口, 大家自己设置.
spring.cloud.stream.bindings.{bing}.destination: 这个是消息沟通的通道, 如果使用了消息分组和分区, 则需要设置该属性, 而且也要一致, 否则消息的接收方接收不到发送方发送的消息.
spring.cloud.stream.bindings.{bing}.group: 当接收方有多个实例在运行的时候, 如果只想消息被处理一次(不管被那个实例处理), 则需要设置这个组, 例如, 在发布方发布一个消息(赵老师让班上3组的一个小朋友今天下课后帮忙把黑板檫了, 那么如果设置了这个组, 意思就是3组的任何一个小朋友檫黑板就行了, 而不用全部去檫).
producer: 这个是为了消息分区设置的, partitionCount指的是分区数.
partitionKeyExpression: 这个是分区表达式, 例如当表达式的值为1, 那么在订阅者的instance-index中为1的接收方, 将会执行该消息.

server:
  port: 8081
spring:
  application:
    name: producer
  cloud:
    stream:
      bindings:
        output1:
          destination: Theme-People #这个需要发布者-订阅者一致才行, 才能在一个通道内
          group: Group-Boy
          content-type: application/json
          producer:
            partitionKeyExpression: headers['partitionKey']
            partitionCount: 2
        output2:
          destination: Theme-People
          group: Group-Girl
          content-type: application/json
          producer: #设置分区键的表达式规则和设置消息分区数量
            partitionKeyExpression: headers['partitionKey']
            partitionCount: 2

3. Person.java

一个vo, 主要是演示如何发送一个类对象

package com.tts.producer.vo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {
    String name;
    int age;
}

3. OutputInterface.java

绑定消息通道, 在spring cloud stream中, 我们通过@input, @output注解来定义消息通道, 改接口可以被@EnableBinding注解的value参数来指定, 从而在应用启动的时候实现对定义消息通道的绑定, 注意Finchley版的spring cloud不允许INTPUT和OUTPUT同名.

package com.tts.producer.message;

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;

@Component
public interface OutputInterface {
    String OUTPUT1 = "output1";
    String OUTPUT2 = "output2";

    @Output(OUTPUT1)
    MessageChannel output1();

    @Output(OUTPUT2)
    MessageChannel output2();
}

5. MessageSender.java

定时发送消息, 通过OUTPUT1, OUTPUT2通道发送消息, 为了演示方便, 我采用了定时器(@Scheduled)定时发送消息.

package com.tts.producer.message;

import com.tts.producer.vo.Person;
import lombok.extern.slf4j.Slf4j;
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.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

@Slf4j
@EnableBinding(value = {OutputInterface.class})
@EnableScheduling
public class MessageSender {
    @Autowired
    OutputInterface outputInterface;

    /**
     * 第一种方法, 没有指定output的MessageChannel, 通过OutputInterface去拿具体的Channel
     * 设置partitionKey主要是为了分区用, 可以根据根据这个partitionKey来分区
     */
    @Scheduled(initialDelay = 1000, fixedRate = 5000)
    public void sendMessageMethod1() {
        Message message = MessageBuilder.withPayload("From sendMessageMethod1")
                .setHeader("partitionKey", 1)
                .build();

        outputInterface.output1().send(message);
    }

    /**
     * 第二种方法, 直接指定output的MessageChannel
     */
    @Autowired
    @Qualifier(OutputInterface.OUTPUT2)
    MessageChannel output;

    @Scheduled(initialDelay = 2000, fixedRate = 4000)
    public void sendMessageMethod2() {
        Person p = new Person();
        p.setName("Person2");
        p.setAge(1);

        output.send(MessageBuilder.withPayload(p)
                .setHeader("partitionKey", 2)
                .build());
    }
}

6: StreamService.java

package com.tts.producer.service;

import com.tts.producer.message.OutputInterface;
import com.tts.producer.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class StreamService {
    @Autowired
    @Qualifier(OutputInterface.OUTPUT1)
    MessageChannel channel1;

    @Autowired
    @Qualifier(OutputInterface.OUTPUT2)
    MessageChannel channel2;

    public void sendMessage(String channel) {
        String message = "Send Message Manually,From Channel:" + channel;
        log.info("Send Message from channel:" + channel);
        switch (channel) {
            case OutputInterface.OUTPUT1:
                channel1.send(MessageBuilder.withPayload(message).build());
                return;
            case OutputInterface.OUTPUT2:
                Person person = new Person("Producer", 1);
                channel2.send(MessageBuilder.withPayload(person).setHeader("partitionKey", 0).build());
                return;
            default:
                log.info("参数错误: " + channel);
                return;
        }
    }
}

7:StreamController.java

手动发送消息 可以通过浏览器直接手动发送消息(localhost:8080/sendMessageByChannalName?channel=output2)

package com.tts.producer.controller;

import com.tts.producer.service.StreamService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class StreamController {
    @Autowired
    StreamService streamService;

    @GetMapping("/sendMessageByChannalName")
    public void sendMessage(String channel) {
        streamService.sendMessage(channel);
    }
}

consumer

1. 引入依赖 (pom.xml)

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.4.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <spring-cloud.version>Finchley.SR1</spring-cloud.version>
</properties>

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

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.1.15</version>
</dependency>

<!--使用aliyun镜像-->
<repositories>
    <repository>
        <id>alimaven</id>
        <name>Maven Aliyun Mirror</name>
        <url>http://maven.aliyun.com/nexus/content/repositories/central/</url>
        <releases>
            <enabled>true</enabled>
        </releases>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

2: application.yml

spring:
  application:
    name: consumer
  cloud:
    stream:
      bindings:
        input1:
          destination: Theme-Animal
          group: Group-Boy
          content-type: application/json
          consumer:
            partitioned: true #是否分区
        input2:
          destination: Theme-People
          group: Group-Girl
          content-type: application/json
          consumer:
            partitioned: true
      instance-index: 0  #用那个分区来接收消息
      instance-count: 2  #分区数

3. InputInterface.java

package com.tts.consumer.message;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

@Component
public interface InputInterface {
    String INPUT1 = "input1";
    String INPUT2 = "input2";

    @Input(INPUT1)
    SubscribableChannel input1();

    @Input(INPUT2)
    SubscribableChannel input2();
}

4. MessageReceiver.java

接收消息类, @StreamListener(InputInterface.INPUT1) 监听INPUT1通道的消息, 然后进行处理

package com.tts.consumer.message;

import com.tts.consumer.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

@Slf4j
@EnableBinding(value = {InputInterface.class})
public class MessageReceiver {
    @StreamListener(InputInterface.INPUT1)
    public void receiveMessageFromChannel1(@Payload Message<String> payload, @Header(AmqpHeaders.CONSUMER_QUEUE) String partition){
        log.info("ReceiveMessageFrom INPUT1, message: {}, Queue:{}", payload.getPayload(), partition);
    }

    @StreamListener(InputInterface.INPUT2)
    public void receiveMessageFromChannel2(Person payload, @Header(AmqpHeaders.CONSUMER_QUEUE) String partition){
        log.info("ReceiveMessageFrom INPUT2, message: {}, Queue:{}", payload, partition);
    }
}

5. Person.java

package com.tts.consumer.vo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {
    String name;
    int age;
}