转载

Spring Cloud Stream初窥

Spring Cloud Stream是一个构建消息驱动的微服务框架。它构建在Spring Boot之上用以创建工业级的应用程序，并且通过Spring Integration提供了和消息代理的连接。Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现(目前仅支持RabbitMQ和Kafka)，同时引入了发布订阅、消费组和分区的语义概念。本文我们就先来看一下Spring Cloud Stream的基本用法。

本文是Spring Cloud系列的第二十九篇文章，了解前二十八篇文章内容有助于更好的理解本文：

1. 使用Spring Cloud搭建服务注册中心

2. 使用Spring Cloud搭建高可用服务注册中心

3. Spring Cloud中服务的发现与消费

4. Eureka中的核心概念

5. 什么是客户端负载均衡

6. Spring RestTemplate中几种常见的请求方式

7. RestTemplate的逆袭之路，从发送请求到负载均衡

8. Spring Cloud中负载均衡器概览

9. Spring Cloud中的负载均衡策略

10. Spring Cloud中的断路器Hystrix

11. Spring Cloud自定义Hystrix请求命令

12. Spring Cloud中Hystrix的服务降级与异常处理

13. Spring Cloud中Hystrix的请求缓存

14. Spring Cloud中Hystrix的请求合并

15. Spring Cloud中Hystrix仪表盘与Turbine集群监控

16. Spring Cloud中声明式服务调用Feign

17. Spring Cloud中Feign的继承特性

18. Spring Cloud中Feign配置详解

19. Spring Cloud中的API网关服务Zuul

20. Spring Cloud Zuul中路由配置细节

21. Spring Cloud Zuul中异常处理细节

22. 分布式配置中心Spring Cloud Config初窥

23. Spring Cloud Config服务端配置细节(一)

24. Spring Cloud Config服务端配置细节(二)之加密解密

25. Spring Cloud Config客户端配置细节

26. Spring Cloud Bus之RabbitMQ初窥

27. Spring Cloud Bus整合RabbitMQ

28. Spring Cloud Bus整合Kafka

本文我们通过一个简单的消息收发效果，来看看Spring Cloud Stream的一个基本使用。

创建工程

首先我们创建一个普通的Spring Boot工程，名为stream-hello，然后添加如下依赖：

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

spring-cloud-starter-stream-rabbit依赖是Spring Cloud Stream对RabbitMQ的封装，这里边也包含了对RabbitMQ的自动化配置，比如连接的RabbitMQ的默认地址就是localhost，默认端口就是5672，默认用户名是guest，默认密码也是guest，由于我们的RabbitMQ都是采用了默认配置，所以这里的配置可以不去修改，一样也可以运行。如果小伙伴需要修改，则和上篇文章一样，直接在application.properties中修改即可。

创建接收器

然后创建一个名为SinkReceiver的类，用来接收RabbitMQ发送来的消息，如下：

@EnableBinding(Sink.class)
public class SinkReceiver {
    private static Logger logger = LoggerFactory.getLogger(StreamHelloApplication.class);
    @StreamListener(Sink.INPUT)
    public void receive(Object playload) {
        logger.info("Received:"+playload);
    }
}

这里我们首先使用了@EnableBinding注解实现对消息通道的绑定，我们在该注解中还传入了一个参数Sink.class，Sink是一个接口，该接口是Spring Cloud Stream中默认实现的对输入消息通道绑定的定义。然后我们在SinkReceiver类中定义了receive方法，并在该方法上添加了@StreamListener注解，该注解表示该方法为消息中间件上数据流的事件监听器，Sink.INPUT参数表示这是input消息通道上的监听处理器。