1. 简介

1.1 官网

【原文】A lightweight event-driven microservices framework to quickly build applications that can connect to external systems. Simple declarative（声名式的） model to send and receive messages using Apache Kafka or RabbitMQ between Spring Boot apps.

【翻译】一个轻量级的事件驱动微服务框架，用于快速构建可连接到外部系统的应用程序。在 Spring Boot 应用程序之间使用 Kafka 或 RabbitMQ 发送和接收消息的简单声明式模型。

1.2 综合

Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。通过使用Spring Cloud Stream，可以有效简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持RabbitMQ 和 Kafka 的自动化配置。

2 程序模型

spring cloud 官网首页中点击相应版本的参考文档。

应用程序的核心部分（Application Core）通过 inputs 与 outputs 管道，与中间件连接，而管道是通过绑定器 Binder 与中间件相绑定的。

3. stream kafka 微服务

3.1 总步骤

(1) 创建生产者步骤

• 导入 spring-cloud-stream-binder-kafka 依赖
• 创建生产者类。在生产者类上添加@EnableBinding()注解，并声明管道
• 定义处理器，在处理器中调用消息生产者，使其发送消息
• 在配置文件中注册 kafka 集群，并指定管道所绑定的主题及类型

(2) 创建消费者步骤

• 导入 spring-cloud-stream-binder-kafka 依赖
• 创建消费者类。在消费者类上添加@EnableBinding()注解，并声明管道。
• 在消费者类中定义消费方法，在方法上添加相应的注解。
• 在配置文件中注册 kafka 集群，并指定管道所绑定的主题

3.2 消息发送给一个主题的生产者

(1) 创建工程 08-stream-kafka-8080
任意复制前面的一个提供者或消费者工程，将其中的除启动类之外的其它代码全部删除。这里复制 02-consumer-8080 工程，并命名为 08-stream-kafka-8080。

修改启动类类名ApplicationStreamKafka8080。

(2) 导入依赖
仅需再添加一个 Spring Cloud Stream Kafka 相关的依赖即可。

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

整个 pom.xml 文件的内容如下：

<groupId>com.abc</groupId>
<artifactId>08-stream-kafka-8080</artifactId>
<version>0.0.1-SNAPSHOT</version><properties><java.version>1.8</java.version><spring-cloud.version>Hoxton.SR1</spring-cloud.version>
</properties><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-kafka</artifactId></dependency><!--actuator依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><!--eureka客户端依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency>
</dependencies><dependencyManagement><dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>${spring-cloud.version}</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins>
</build>

(3) 创建生产者类

@Component
// 将MQ与生产者类通过消息管道相绑定
// org.springframework.cloud.stream.messaging.Source
@EnableBinding(Source.class)
public class SomeProducer {// 必须使用byName方式的自动注入// 系统中还定义了名称为nullChannel和errorChannel的两个同类型管道@Autowired@Qualifier(Source.OUTPUT)//Source.OUTPUT="output"private MessageChannel channel;public String sendMessage(String msg) {// 通过消息管道发送消息，即将消息写入到消息管道，再通过消息管道写入到MQchannel.send(MessageBuilder.withPayload(msg).build());return msg;}
}

(4) 创建处理器

@RestController
public class SomeController {// 将生产者注入@Autowiredprivate SomeProducer producer;@PostMapping("/msg/send")public String sendHandler(@RequestParam("message") String msg) {// 生产者发送消息return producer.sendMessage(msg);}
}

(5) 创建配置文件

spring:application:name: abcmsc-consumer-departcloud:stream:kafka:binder:# 指定要连接的kafka集群brokers: kafkaOS1:9092,kafkaOS2:9092,kafkaOS3:9092# 指定是否自动创建主题auto-create-topics: truebindings:# 指定要绑定的输出管道，及要输出到单管道中的消息主题及类型output:destination: persons # 主题content-type: text/plaineureka:client:service-url:defaultZone: http://localhost:8000/eureka

(6) 演示
kafka消费者先用命令行方式：
因为我机器上已经有persons主题了，为了演示自动创建主题，换成names：

发送多次：

3.3 消息发送给多个主题的生产者

前面工程中通过 Source 的 MessageChannel 完成了将消息发送给某个指定主题的功能，若要将消息发送给多个主题，则需要自定义 Channel。

(1) 创建工程 08-stream-kafka2-8080
复制前面的 08-stream-kafka-8080，重命名为 08-stream-kafka2-8080。
修改启动类类名ApplicationStreamKafka28080。

(2) 定义 Source 接口
模拟 Source 接口自定义一个 Source 接口。

/*** 自定义管道*/
public interface CustomSource {//管道名称String CHANNEL_NAME = "xxx";@Output(CustomSource.CHANNEL_NAME)MessageChannel output();
}

(3) 修改发布者类

@Component
// 将MQ与生产者类通过消息管道相绑定
@EnableBinding({Source.class, CustomSource.class})
public class SomeProducer {// 必须使用byName方式的自动注入@Autowired@Qualifier(Source.OUTPUT)private MessageChannel channel;@Autowired@Qualifier(CustomSource.CHANNEL_NAME)private MessageChannel customChannel;public String sendMessage(String msg) {// 将消息写入到两个管道，将会写入到两个主题channel.send(MessageBuilder.withPayload(msg).build());customChannel.send(MessageBuilder.withPayload(msg).build());return msg;}
}

(4) 修改配置文件
在配置文件中添加如下输出目标。

spring:application:name: abcmsc-consumer-departcloud:stream:kafka:binder:# 指定要连接的kafka集群brokers: kafkaOS1:9092,kafkaOS2:9092,kafkaOS3:9092# 指定是否自动创建主题auto-create-topics: truebindings:# 指定要绑定的输出管道，及要输出到单管道中的消息主题及类型output:destination: namescontent-type: text/plainxxx:destination: citiescontent-type: text/plaineureka:client:service-url:defaultZone: http://localhost:8000/eureka

(5) 效果
发送消息：dddddd、eeeeee、fffff

结果：

3.4 创建消息消费者 – @PostConstruct 方式

Spring Cloud Stream 提供了三种创建消费者的方式，这三种方式的都是在消费者类的“消费”方法上添加注解。只要有新的消息写入到了管道，该“消费”方法就会执行。只不过三种注解，其底层的实现方式不同。即当新消息到来后，触发“消费”方法去执行的实现方式不同。

• @PostConstruct：以发布/订阅方式实现
• @ServiceActivator：新的消息激活服务方式实现
• @StreamListener：以监听方式实现

为了简单起见，这里就不再单独创建消费者工程了，都直接定义在前面的生产者工程中。

(1) 创建消费者类

@Component
@EnableBinding(Sink.class) //和生产者Source注解类似，绑定了一个输入通道
public class SomeConsumer {@Autowired@Qualifier(Sink.INPUT)private SubscribableChannel channel;@PostConstruct       //以发布/订阅方式实现public void printMessage() {//为输入通道注册消息处理程序。channel.subscribe(msg -> {// MessageHeaders headers = msg.getHeaders();System.out.println(new String((byte[])msg.getPayload()));});}
}

(2) 修改配置文件

spring:application:name: abcmsc-consumer-departcloud:stream:kafka:binder:# 指定要连接的kafka集群brokers: kafkaOS1:9092,kafkaOS2:9092,kafkaOS3:9092# 指定是否自动创建主题auto-create-topics: truebindings:# 指定要绑定的输出管道，及要输出到单管道中的消息主题及类型output:destination: namescontent-type: text/plainxxx:destination: citiescontent-type: text/plain# 指定要绑定的输入管道，及要消费的管道中的消息主题input:destination: nameseureka:client:service-url:defaultZone: http://localhost:8000/eureka

(3) 效果

3.5 创建消息消费者 – @ServiceActivator 方式

该注解所标注的方法是以服务的形式出现的，只要管道中的数据发生了变化就会激活该服务。

(1) 注释掉前面的消费者
将前面的消费者类注释掉，其将不会出现在 Spring 容器。

(2) 创建消费者类

@Component
@EnableBinding(Sink.class)//绑定一个输入通道
public class SomeConsumer2 {//指示方法能够处理消息或消息有效负载。//注释为@Serviceactivator的方法可以接受类型//为{@link org.springframework.messaging.Message}或预期消息有效负载类型的参数。//如果需要，{@link org.springframework.beans.SimpleTypeConverter}支持的任何类型//转换都将应用于消息有效负载。通过使//用{@link org.springframework.messaging.handler.annotation.Header @Header}参数注解//消息头的值也可以作为消息参数传递。//注释方法的返回值可以是任何类型。如果返回值不是消息，则将以该对象作为其有效负载创建应答消息。@ServiceActivator(inputChannel = Sink.INPUT)public void printMessage(Object msg) {System.out.println(msg);}
}

3.6 创建消息消费者 – @SteamListener 方式

该方式是以监听的方式实现，只要管道中的流数据发生变化，其就会触发该注解所标注的方法的执行。

(1) 注释掉前面的消费者

(2) 创建消费者类

@Component
@EnableBinding(Sink.class)//绑定一个输入通道
public class SomeConsumer3 {//将一个方法标记为通过{@link EnableBinding}(例如通道)声明的输入的侦听器的注解。@StreamListener(Sink.INPUT)public void printMessage(Object msg) {System.out.println(msg);}
}