Spring Cloud集成RabbitMQ的使用
2024-09-08 01:37:32
同步 or 异步
前言：我们现在有一个用微服务架构模式开发的系统，系统里有一个商品服务和订单服务，且它们都是同步通信的。
目前我们商品服务和订单服务之间的通信方式是同步的，当业务扩大之后，如果还继续使用同步的方式进行服务之间的通信，会使得服务之间的耦合增大。例如我们登录操作可能需要同步调用用户服务、积分服务、短信服务等等，而服务之间可能又依赖别的服务，那么这样一个登录过程就会耗费不少的时间，以致用户的体验降低。
那我们在微服务架构下要如何对服务之间的通信进行解耦呢？这就需要使用到消息中间件了，消息中间件可以帮助我们将同步的通信转化为异步通信，服务之间只需要对消息队列进行消息的发布、订阅即可，从而解耦服务之间的通信依赖。
目前较为主流的消息中间件：
- RabbitMQ
- Kafka
- ActiveMQ
异步通信特点：
- 客户端请求不会阻塞进程，服务端的响应可以是非即时的
异步的常见形态：
- 推送通知
- 请求/异步响应
- 消息队列
MQ应用场景：
- 异步处理
- 流量削峰
- 日志处理
- 应用解耦
更多关于消息中间件的描述，可以参考我另一篇文章：
RabbitMQ的基本使用
在上文 Spring Cloud Config - 统一配置中心 中，已经演示过使用Docker安装RabbitMQ，所以这里就不再浪费篇幅演示了。
直接进入正题，我们以订单服务和商品服务示例，首先在订单服务的项目中，加入mq的依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
在配置文件中增加RabbitMQ的相关配置项：
到订单服务的项目中，新建一个message包，在该包中创建一个MqReceiver类，我们来看看RabbitMQ的基本操作。代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @program: sell_order
* @description: 接收消息，即消费者
* @author: 01
* @create: 2018-08-21 22:24
**/
@Slf4j
@Component
public class MqReceiver {
/**
* 接收消息并打印
*
* @param message message
*/
@RabbitListener(queues = "myQueue")
public void process(String message) {
// @RabbitListener注解用于监听RabbitMQ，queues指定监听哪个队列
log.info(message);
}
}
因为RabbitMQ上还没有myQueue这个队列，所以我们还得到RabbitMQ的管理界面上，创建这个队列，如下：
然后新建一个测试类，用于发送消息到队列中，代码如下：
package org.zero.springcloud.order.server;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @program: sell_order
* @description: 发送消息，即消息发布者
* @author: 01
* @create: 2018-08-21 22:28
**/
@RunWith(SpringRunner.class)
@SpringBootTest
public class MqSenderTest {
@Autowired
private AmqpTemplate amqpTemplate;
@Test
public void send() {
for (int i = 0; i < 100; i++) {
amqpTemplate.convertAndSend("myQueue", "第" + i + "条消息");
}
}
}
运行该测试类，运行成功后到OrderApplication的控制台上，看看是否接收并打印了接收到的消息。正常情况应如下：
基本的消费者和发布者的代码我们都已经编写过，并且也测试成功了。但有个小问题，我们要监听一个不存在的队列时，需要手动去新建这个队列，感觉每次都手动新建挺麻烦的。有没有办法当队列不存在时，自动创建该队列呢？答案是有的，依旧使用之前的那个注解，只不过这次的参数要换成queuesToDeclare。示例代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @program: sell_order
* @description: 接收消息，即消费者
* @author: 01
* @create: 2018-08-21 22:24
**/
@Slf4j
@Component
public class MqReceiver {
/**
* 接收并打印消息
* 可以当队列不存在时自动创建队列
*
* @param message message
*/
@RabbitListener(queuesToDeclare = @Queue("myQueue"))
public void process2(String message) {
// @RabbitListener注解用于监听RabbitMQ，queuesToDeclare可以创建指定的队列
log.info(message);
}
}
以上我们通过示例简单的介绍了消息的收发及队列的创建，本小节则介绍一下exchange 的自动绑定方式。当需要自动绑定 exchange 时，我们也可以通过 bindings 参数完成。示例代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @program: sell_order
* @description: 接收消息，即消费者
* @author: 01
* @create: 2018-08-21 22:24
**/
@Slf4j
@Component
public class MqReceiver {
/**
* 接收并打印消息
* 可以当队列不存在时自动创建队列，以及自动绑定指定的Exchange
* @param message message
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue("myQueue"),
exchange = @Exchange("myExchange")
))
public void process3(String message) {
// @RabbitListener注解用于监听RabbitMQ，bindings可以创建指定的队列及自动绑定Exchange
log.info(message);
}
}
消息分组我们也是可以通过 bindings 参数完成，例如现在有一个数码供应商服务和一个水果供应商服务，它们都监听着同一个订单服务的消息队列。但我希望数码订单的消息被数码供应商服务消费，而水果订单的消息被水果供应商服务消费。所以我们就需要用到消息分组。示例代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @program: sell_order
* @description: 接收消息，即消费者
* @author: 01
* @create: 2018-08-21 22:24
**/
@Slf4j
@Component
public class MqReceiver {
/**
* 数码供应商服务 - 接收消息
*
* @param message message
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue("computerOrder"),
exchange = @Exchange("myOrder"),
key = "computer" // 指定路由的key
))
public void processComputer(String message) {
log.info("computer message : {}", message);
}
/**
* 水果供应商服务 - 接收消息
*
* @param message message
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue("computerOrder"),
exchange = @Exchange("myOrder"),
key = "fruit" // 指定路由的key
))
public void processFruit(String message) {
log.info("fruit message : {}", message);
}
}
测试代码如下，通过指定key进行消息的分组，将消息发送到数码供应商服务：
package org.zero.springcloud.order.server;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @program: sell_order
* @description: 发送消息，即消息发布者
* @author: 01
* @create: 2018-08-21 22:28
**/
@RunWith(SpringRunner.class)
@SpringBootTest
public class MqSenderTest {
@Autowired
private AmqpTemplate amqpTemplate;
@Test
public void sendOrder() {
for (int i = 0; i < 100; i++) {
// 第一个参数指定队列，第二个参数来指定路由的key，第三个参数指定消息
amqpTemplate.convertAndSend("myOrder", "computer", "第" + i + "条消息");
}
}
}
重启项目后，运行以上测试代码，控制台输出如下，可以看到只有数码供应商服务才能够接收到消息，而水果供应商服务是接收不到的。这就完成了消息分组：
Spring Cloud Stream的使用
Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于Spring Boot 来创建独立的，可用于生产的Spring 应用程序。他通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前仅支持RabbitMQ、Kafka。
什么是Spring Integration ？ Integration 集成
企业应用集成（EAI）是集成应用之间数据和服务的一种应用技术。四种集成风格：
- 文件传输：两个系统生成文件，文件的有效负载就是由另一个系统处理的消息。该类风格的例子之一是针对文件轮询目录或FTP目录，并处理该文件。
- 共享数据库：两个系统查询同一个数据库以获取要传递的数据。一个例子是你部署了两个EAR应用，它们的实体类（JPA、Hibernate等）共用同一个表。
- 远程过程调用：两个系统都暴露另一个能调用的服务。该类例子有EJB服务，或SOAP和REST服务。
- 消息：两个系统连接到一个公用的消息系统，互相交换数据，并利用消息调用行为。该风格的例子就是众所周知的中心辐射式的（hub-and-spoke）JMS架构。
Spring Integration作为一种企业级集成框架，遵从现代经典书籍《企业集成模式》，为开发者提供了一种便捷的实现模式。Spring Integration构建在Spring控制反转设计模式之上，抽象了消息源和目标，利用消息传送和消息操作来集成应用环境下的各种组件。消息和集成关注点都被框架处理，所以业务组件能更好地与基础设施隔离，从而降低开发者所要面对的复杂的集成职责。
模型图：
现在我们来看看Spring Cloud Stream的基本使用，到订单服务项目上，增加如下依赖：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
然后是在配置文件中，配置rabbitmq的相关信息，只不过我们之前已经配置过了所以不用配置了。
我们来看看如何使用Spring Cloud Stream发送和接收消息，首先创建一个接口，定义input和output方法。代码如下：
package org.zero.springcloud.order.server.message;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
public interface StreamClient {
// 接收消息、入口
@Input("myMessageInput")
SubscribableChannel input();
// 发送消息、
@Output("myMessageOutput")
MessageChannel output();
}
创建一个消息接收者。代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
/**
* @program: sell_order
* @description: 消息接收者
* @author: 01
* @create: 2018-08-22 22:16
**/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
@StreamListener("myMessageOutput")
public void process(String message) {
log.info("message : {}", message);
}
}
消息发送者，这里作为一个Controller存在。代码如下：
package org.zero.springcloud.order.server.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.zero.springcloud.order.server.message.StreamClient;
/**
* @program: sell_order
* @description: 消息发送者
* @author: 01
* @create: 2018-08-22 22:18
**/
@RestController
public class SendMessageController {
private final StreamClient streamClient;
@Autowired
public SendMessageController(StreamClient streamClient) {
this.streamClient = streamClient;
}
@GetMapping("/send/msg")
public void send() {
for (int i = 0; i < 100; i++) {
MessageBuilder<String> messageBuilder = MessageBuilder.withPayload("这是第" + i + "条消息");
streamClient.output().send(messageBuilder.build());
}
}
}
因为我们的微服务可能会部署多个实例，若有多个实例需要对消息进行分组，否则所有的服务实例都会接收到相同的消息。在配置文件中，增加如下配置完成消息的分组：
spring:
...
cloud:
...
stream:
bindings:
myMessageOutput:
group: order
...
重启项目，访问http://localhost:9080/send/msg，控制台输出如下：
注：Spring Cloud Stream可以在项目启动的时候自动创建队列，在项目关闭的时候自动删除队列
在实际的开发中，我们一般发送的消息通常会是一个java对象而不是字符串。所以我们来看看如何发送对象，其实和发送字符串几乎是一样的。消息发送者代码如下：
package org.zero.springcloud.order.server.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.zero.springcloud.order.server.dto.OrderDTO;
import org.zero.springcloud.order.server.message.StreamClient;
/**
* @program: sell_order
* @description: 消息发送者
* @author: 01
* @create: 2018-08-22 22:18
**/
@RestController
public class SendMessageController {
private final StreamClient streamClient;
@Autowired
public SendMessageController(StreamClient streamClient) {
this.streamClient = streamClient;
}
/**
* 发送OrderDTO对象
*/
@GetMapping("/send/msg")
public void send() {
OrderDTO orderDTO = new OrderDTO();
orderDTO.setOrderId("123465");
MessageBuilder<OrderDTO> messageBuilder = MessageBuilder.withPayload(orderDTO);
streamClient.output().send(messageBuilder.build());
}
}
消息接收者也只需要在方法参数上声明这个对象的类型即可。代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
import org.zero.springcloud.order.server.dto.OrderDTO;
/**
* @program: sell_order
* @description: 消息接收者
* @author: 01
* @create: 2018-08-22 22:16
**/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
/**
* 接收OrderDTO对象
* @param message message
*/
@StreamListener("myMessageOutput")
public void process(OrderDTO message) {
log.info("message : {}", message);
}
}
另外需要提到的一点是，默认情况下，java对象在消息队列中是以base64编码存在的，我们也都知道base64不可读。为了方便查看堆积在消息队列里的对象数据，我们希望java对象是以json格式的字符串呈现，这样就方便我们人类阅读。至于这个问题，我们只需要在配置文件中，增加一段content-type的配置即可。如下：
spring:
...
cloud:
...
stream:
bindings:
myMessageOutput:
group: order
content-type: application/json
...
重启项目，访问http://localhost:9080/send/msg，控制台输出如下：
2018-08-22 23:32:33.704 INFO 12436 --- [nio-9080-exec-4] o.z.s.o.server.message.StreamReceiver
: message : OrderDTO(orderId=123465, buyerName=null, buyerPhone=null, buyerAddress=null, buyerOpenid=null,
orderAmount=null, orderStatus=null, payStatus=null, createTime=null, updateTime=null, orderDetailList=null)
当我们接收到消息的时候，可能会需要返回一段特定的消息，表示消息已收到之类的。至于这个功能，我们通过@SendTo注解即可完成。代码如下：
package org.zero.springcloud.order.server.message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
import org.zero.springcloud.order.server.dto.OrderDTO;
/**
* @program: sell_order
* @description: 消息接收者
* @author: 01
* @create: 2018-08-22 22:16
**/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
/**
* 接收OrderDTO对象
* @param message message
*/
@StreamListener("myMessageOutput")
@SendTo("myMessageInput")
public String process(OrderDTO message) {
log.info("message : {}", message);
return "success";
}
@StreamListener("myMessageInput")
public void success(String message) {
log.info("message : {}", message);
}
}
重启项目，访问http://localhost:9080/send/msg，控制台输出如下：
Spring Cloud Stream 再一次简化了我们在分布式环境下对消息中间件的操作，配置好消息中间件的连接地址及用户密码后，在开发的过程中，我们只需要关注input和output，对消息中间件的操作基本是无感知的。
最新文章
- Mac上的欧路词典单词本迁移到有道词典 #node.js
- schedule() 和 scheduleAtFixedRate() 区别
- C#程序设计基础——字符串
- Keil中Memory Model和Code Rom Size说明
- linux下串口调试工具/串口终端推荐: picocom
- 有意思的cmd命令
- Nginx1.8.0版本平滑升级新版本1.9.7
- PHP xdebug的安装
- 恶补web之一:html学习(2)
- 唱吧DevOps的落地，微服务CI/CD的范本技术解读
- MySQL使用细节
- CMSC 216 Exercise #5
- 剑指Offer 57. 二叉树的下一个结点 （二叉树）
- .net C# 利用Session防重复点击防重复提交
- javascript UI lib
- java笔记--问题总结
- C C++ 去除 unused的提示
- nginx端合并JS
- python 获取中文文件名的输出
- Inno Setup 软件封装