Spring Cloud Stream是一个构建消息驱动的微服务框架。它构建在Spring Boot之上用以创建工业级的应用程序,并且通过Spring Integration提供了和消息代理的连接。Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现(目前仅支持RabbitMQ和Kafka),同时引入了发布订阅、消费组和分区的语义概念。本文我们就先来看一下Spring Cloud Stream的基本用法。
本文我们通过一个简单的消息收发效果,来看看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消息通道上的监听处理器。
Ok,做好上面的操作之后,我们就可以启动我们的项目了,在项目的启动日志中,我们可以看到如下内容:
这个表示我们的工程已经通过guest用户创建了一个指向RabbitMQ的连接。此时,在浏览器中打开RabbitMQ的管理页面,一样也可以看到连接,如下:
图中红线部分就是我们新建的队列,点击该队列,找到Publish message选项,我们可以从这里给我们的项目发送一条消息,如下:
此时我们可以在项目的控制台中看到消息已经收到了,如下:
只是消息没有序列化而已。
好了,Spring Cloud Stream的初步使用我们就先介绍到这里,有问题欢迎留言讨论。
参考资料:
1.《Spring Cloud微服务》