7.SpringBoot整合RabbitMQ实现微服务间的异步消息沟通

1.需求：

前我们已经开发了两个微服务，一个用来管理设备目录，一个用来管理技术状态。现在需要通过MQ来沟通彼此。两个微服务之间的关系如图：

左右都是一个单独的微服务，彼此之间没有直接的关联关系，各自的信息是分库保存的。也就是说在通过设备目录中存在的信息建立技术状态时，技术状态中保存的并非设备目录信息的引用，而是副本。这样可以保证即使在设备目录服务无法工作时技术状态服务不受影响。由于保存的是副本而非引用，那么当设备目录中的信息发生变化时，就需要使用消息队列发送通知如：id为1的设备名称变为‘发电机’这样的消息给技术状态服务，技术状态服务接收通知后，根据消息内容更新自身数据库中的信息，保证整个系统中数据的一致性。

2.搭建RabbitMQ服务器

依然使用Docker的方式来部署RabbitMQ服务器。

按照RabbitMQ官方Docker使用说明

先使用docker pull rabbitmq命令获取RabbitMQ的镜像。

接着使用

docker run -d --hostname my-rabbit --name some-rabbit
-p 8080:15672
-e RABBITMQ_DEFAULT_USER=user -e RABBITMQ_DEFAULT_PASS=password
rabbitmq:3-management

测试访问控制台

部署一个指定用户名和密码并且可以通过8080端口访问控制台的RabbitMQ服务器。

3.配置SpringBoot

在项目的POM文件中加入Spring-AOP和Spring-AMQP的依赖

<!-- AOP -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<!-- rabbitMQ -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

通过SpringBoot配置文件整合RabbitMQ

#rabbitMQ
spring.rabbitmq.host=192.168.227.101
spring.rabbitmq.port=5672
spring.rabbitmq.username=user
spring.rabbitmq.password=password

4.发送消息

思路是这样的，当设备目录服务的设备信息变更方法正常执行完成后，我们使用AOP触发消息的发送

package an.aop;

import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSON;
import an.entity.Catalog;
//声明切面
@Component
@Aspect
public class CatalogAop {
//在引入Spring-AMQP后可以直接使用AmqpTemplate
@Autowired
private AmqpTemplate rabbitTemplate;

//使用@AfterReturning并声明切点
@AfterReturning(value="execution(* an.catalog.CatalogService.editCatalog(..))" , returning="result")
public void doAccessCheck(JoinPoint joinPoint , Object result) {
//组装要发送的消息内容
Object[] args = joinPoint.getArgs();
Long catalogId = (Long) args[0];
String name = (String) args[1];
String model = (String) args[2];
String unit = (String) args[3];
Boolean unique = (Boolean) args[4];
Boolean parent = (Boolean) args[5];
Catalog catalog = new Catalog(catalogId, name, model, unit, unique, parent);
//由于editCatalog正常执行的返回值是void，所以如果返回为null说明成功执行，可以开始发送消息，如果失败则什么都不发送
if(result == null) {
//使用已封装好的convertAndSend(String exchange , String routingKey , Object message)使用特定的routingKey发送消息到指定的exchange
rabbitTemplate.convertAndSend("catalogExchange" , "editCatalogKey" , JSON.toJSONString(catalog));
}
}
}

5.接收消息

package an.aop;

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.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSON;
import an.entity.Catalog;
import an.techstatus.TechStatusRepository;

@Component
public class Receiver {
//注入技术状态数据持久层
@Autowired
private TechStatusRepository repository;

//使用@RabbitListener监听指定队列、指定exchange、指定routingKey的消息
//同时@RabbitListener有建立队列、exchange、routingKey的功能
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(value = "editCatalog" , durable = "true") ,
exchange = @Exchange(value = "catalogExchange" , type = "topic" , durable = "true") ,
key = "editCatalogKey")
)
public void receiveMessage(String message) throws Exception {
Catalog catalog = JSON.parseObject(message , Catalog.class);
//进行更新
repository.updateDeviceInfo(catalog);
}
}

6.启动服务后验证RabbitMQ中队列、exchange、routingKey的建立

7.验证