【架构系列】RabbitMQ应用场景及在实际项目中如何搭建可靠的RabbitMQ架构体 ...

作者:后端小肥肠
  
  创作不易，未经允许克制转载。
  1. 前言


RabbitMQ，作为一款高性能、可靠的消息队列软件，已经成为很多企业和开辟团队的首选之一。它的灵活性和可扩展性使得它适用于各种应用场景，从简朴的使命队列到复杂的分布式系统。本文将深入探讨RabbitMQ的应用场景以及如何在实际项目中构建可靠的RabbitMQ架构体系。
2. RabbitMQ应用场景

2.1 异步处理

在现代应用中，异步消息处理是提升用户体验和系统服从的关键。RabbitMQ可以有效地用于多种异步处理使命，比方：

• 用户注册后的邮件发送：用户注册后，通过RabbitMQ发送一个消息到队列中，由背景服务监听并处剃头送邮件的使命，从而不会延迟用户的注册过程。
  
• 订单处理：在电商平台中，订单处理包括库存管理、支付确认等多个步骤，RabbitMQ可以用来在这些服务间异步转达订单信息，确保处理流程的连续性和服从。
  

2.2 应用解耦

RabbitMQ支持多种通信模式，如点对点、发布/订阅等，这些模式帮助系统各部门保持低耦合度，便于独立扩展和维护。比方：

• 微服务架构中的服务通信：在微服务架构中，RabbitMQ允许各个微服务之间通过消息进行交互，而不是直接调用对方的API，这种方式减少了服务间的直接依赖。
  

2.3 流量削峰

在流量高峰期，如促销或大型活动期间，系统大概会遭遇巨大的访问压力。RabbitMQ可以用来缓冲入站消息，如订单或请求，从而保护后端服务不被过载：

• 秒杀活动中的订单处理：在秒杀活动中，大量的购买请求可以先进入RabbitMQ队列，系统根据处理本领逐步从队列中取出并处理这些请求，有效避免了系统瓦解。
  

2.4 通信与集成

RabbitMQ提供了一个灵活的消息转达系统，可以集成复杂的企业系统。它支持多种协媾和广泛的开辟语言库，适用于：

• 跨平台通信：在不同操作系统和不同编程语言编写的应用之间，RabbitMQ可以作为消息转达中间件，实现这些系统的有效通信。
  

2.5 日记处理和应用监控

RabbitMQ也常用于系统日记处理和监控。它可以聚合各服务产生的日记信息，并传输到日记分析系统：

• 集中式日记管理：通过RabbitMQ，各个系统和应用的日记可以被统一网络至一个中心处理位置，便于进行日记分析、监控和报警。
  

2.6 数据同步

RabbitMQ 在数据同步中扮演着告急的角色，特殊是在分布式系统中，它能够确保数据在多个系统或组件之间保持同等性和最新状态。这对于维护数据的完整性和实时性至关告急。比方：

• 数据库同步：在多地数据中心运营的情况下，RabbitMQ 可以用来同步不同地点的数据库。通过消息队列，当一个数据中心的数据库更新时，相应的变更可以通过 RabbitMQ 发送到其他数据中心，从而包管所有地点的数据同等。
  
• 实时数据复制：在金融服务或电子商务平台，实时数据复制是包管高可用性和劫难恢复的关键。使用 RabbitMQ，可以实现高效的数据复制战略，如将交易数据从主系统复制到备份系统或分析数据库。
  
• 缓存刷新：在使用缓存进步应用性能的情况下，RabbitMQ 可以用来在数据更新时主动通知系统刷新缓存。如许，用户总是能够获取到最新的数据，而不是过时的缓存数据。
  

通过这些应用场景，可以看出RabbitMQ在现代软件架构中扮演的多样化角色，不但增强了系统的可靠性和伸缩性，还进步了开辟和运维的服从。
3. 在项目中如何搭建稳定RabbitMQ架构体系

3.1. RabbitMQ安装

网上RabbitMQ安装教程很多，本文只简述基于docker安装的核心步骤：
1. 环境准备，准备Cenos虚拟机，我的是7.x版本：

2. 拉取或解压RabbitMQ镜像：

3. 运行docker容器：

1. docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -v /home/docker/rabbitmq/rabbitmq:/var/lib/rabbitmq -v /home/docker/rabbitmq/rabbitmq_conf:/etc/rabbitmq   -e RABBITMQ_DEFAULT_VHOST=km_vhost  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:latest

复制代码

4. 进入容器 ：

1. docker exec -it 容器id /bin/bash

复制代码

5. 运行rabbitmq-plugins enable rabbitmq_management（解决无法访问网页端15672端口题目），即可完成RabbitMQ安装。

3.2. 总体技能流程

本文以异步处理应用场景为例，展示如何构建稳定可靠的RabbitMQ架构体系：

上述流程为异步消息通信的技能流程，在异步消息通信中当消息投递后就立刻返回了效果，我们无法获取消息消耗的具体过程，这就导致了固然我们可以马上获取程序返回状态，但是程序实行细节或是否失败无法通过程序相应返回的方式获取。
基于以上RabbitMQ异步通信的优缺点，我们要搭建一个可靠的RabbitMQ架构需要从以下几个方面入手：
生产者稳定架构：
1. 消息投递回调监听。创建消息投递回调监听函数，监听生产者投递的消息是否投递乐成。
2. 消息确认表创建。创建消息确认表（message_confirmation），记录消息投递状态，其中字段status反应了是否投递乐成（0为为投递乐成，1为投递乐成）。

1. CREATE TABLE "public"."message_confirmation" (
2.   "id" varchar(100) COLLATE "pg_catalog"."default" NOT NULL,
3.   "status" int4,
4.   "create_time" timestamp(6),
5.   "update_time" timestamp(6),
6.   "message" varchar(255) COLLATE "pg_catalog"."default",
7.   CONSTRAINT "message_confirmation_pkey" PRIMARY KEY ("id")
8. )
9. ;
11. ALTER TABLE "public"."message_confirmation"
12.   OWNER TO "postgres";

复制代码

3. 创建定时使命监听消息投递确认表。每隔一段时间遍历消息确认表，筛选出status为0的消息数据，进行重复投递动作。
消耗者稳定架构
1. 死信队列运用。由于网络或外部因素导致消息消耗失败，可将消息投递至死信队列进行二次消耗。
2. 日记表记录。如死信队列也消耗失败，可将消息写入日记表（message_error）后进行手动消耗，由技能人员获取日记表中消耗失败记录，排查消耗失败缘故起因。

1. CREATE TABLE "public"."message_error" (
2.   "id" varchar(100) COLLATE "pg_catalog"."default" NOT NULL,
3.   "message_id" varchar(100) COLLATE "pg_catalog"."default" NOT NULL,
4.   "error_log" text COLLATE "pg_catalog"."default",
5.   "create_time" timestamp(6),
6.   "update_time" timestamp(6),
7.   CONSTRAINT "message_error_pkey" PRIMARY KEY ("id")
8. )
9. ;
11. ALTER TABLE "public"."message_error"
12.   OWNER TO "postgres";

复制代码

3.3. 实战讲解

3.3.1. 环境配置

3.3.1.1. 所需版本工具

依赖版本Spring Boot2.6.3Java1.8以上postgresSQL13.12RabbitMQ3.9.11 3.3.1.2. pom依赖

1. <dependencies>
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter-web</artifactId>
5.         </dependency>
6.         <dependency>
7.             <groupId>com.baomidou</groupId>
8.             <artifactId>mybatis-plus-boot-starter</artifactId>
9.         </dependency>
10.         <dependency>
11.             <groupId>org.projectlombok</groupId>
12.             <artifactId>lombok</artifactId>
13.         </dependency>
14.         <dependency>
15.             <groupId>org.springframework.boot</groupId>
16.             <artifactId>spring-boot-starter-validation</artifactId>
17.         </dependency>
18.          <dependency>
19.             <groupId>org.springframework.boot</groupId>
20.             <artifactId>spring-boot-starter-amqp</artifactId>
21.         </dependency>
22.        <dependency>
23.             <groupId>org.postgresql</groupId>
24.             <artifactId>postgresql</artifactId>
25.         </dependency>
26. </dependencies>

复制代码

3.3.2. 生产者核心代码讲解

3.3.2.1. yml配置

1. server:
2.   port: 8873
3. spring:
4.   datasource:
5.     url: jdbc:postgresql://127.0.0.1:5432/xfc_mq_producer
6.     username: postgres
7.     password: postgres
8.     driver-class-name: org.postgresql.Driver
9.   rabbitmq:
10.     port: 5672
11.     host: 192.168.10.11
12.     username: admin
13.     password: admin
14.     virtual-host: my_vhost
15.     publisher-confirm-type: correlated
16.     listener:
17.       simple:
18.         acknowledge-mode: manual

复制代码

3.3.2.2. 编写回调函数

1. @PostConstruct
2.     public void regCallback() {
3.         // 消息发送成功以后，给予生产者的消息回执,来确保生产者的可靠性
4.         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
5.             @Override
6.             public void confirm(CorrelationData correlationData, boolean ack, String cause) {
7.                 log.info("cause:"+cause);
8.                 // 如果ack为true代表消息已经收到
9.                 String messageId = correlationData.getId();
11.                 if (!ack) {
12.                     // 这里可能要进行其他的方式进行存储
13.                     log.error("MQ队列应答失败，messageId是:" + messageId);
14.                     return;
15.                 }
17.                 try {
18.                     MessageConfirmation messageConfirmation = messageConfirmationMapper.selectById(messageId);
19.                     messageConfirmation.setStatus(1);
20.                     int count=messageConfirmationMapper.updateById(messageConfirmation);
21.                     if (count == 1) {
22.                         log.info("本地消息状态修改成功，消息成功投递到消息队列中...");
23.                     }
24.                 } catch (Exception ex) {
25.                     log.error("本地消息状态修改失败，出现异常：" + ex.getMessage());
26.                 }
27.             }
28.         });
29.     }

复制代码

上述回调函数主要用于监听生产者发送的消息是否发送乐成，并将消息发送状态更新至消息确认表中。
3.3.2.3. 编写定时使命监听消息确认表

1. @Configuration
2. @EnableScheduling
3. @Slf4j
4. public class confirmMessageTaskService {
5.     @Autowired
6.     private RabbitTemplate rabbitTemplate;
8.     @Autowired
9.     MessageConfirmationMapper messageConfirmationMapper;
11.     @Scheduled(cron = "0 */1 * * * ?")
12.     public void sendMessage(){
13.         // 把消息为0的状态消息重新查询出来，投递到MQ中。
14.         LambdaQueryWrapper<MessageConfirmation> queryWrapper = new LambdaQueryWrapper<>();
15.         queryWrapper.eq(MessageConfirmation::getStatus, 0);
16.         List<MessageConfirmation> noConfirmMessages = messageConfirmationMapper.selectList(queryWrapper)
17.                 .stream()
18.                 .collect(Collectors.toList());
19.         noConfirmMessages.forEach((noConfirmMessage)->{
20.             rabbitTemplate.convertAndSend("xz_push_exchange","", JsonUtil.obj2String(noConfirmMessage),
21.                     new CorrelationData(noConfirmMessage.getId()));
22.         });
23.     }
24. }

复制代码

 上述定时使命为每分钟遍历消息确认表，将status=0的消息筛选出来进行消息投递。
3.3.2.4. 消息投递

1.     public void sendMessage(MessageConfirmation messageConfirmation) {
2.         messageConfirmationMapper.insert(messageConfirmation);
3.         rabbitTemplate.convertAndSend("xfc_fanout_exchange","", JsonUtil.obj2String(messageConfirmation),
4.                 new CorrelationData(messageConfirmation.getId()));
5.     }

复制代码

3.4. 消耗者核心代码讲解

3.4.1. yml配置

1. server:
2.   port: 8872
3. spring:
4.   datasource:
5.     url: jdbc:postgresql://127.0.0.1:5432/xfc_mq_consumer
6.     username: postgres
7.     password: postgres
8.     driver-class-name: org.postgresql.Driver
9.   rabbitmq:
10.     port: 5672
11.     host: 192.168.10.11
12.     username: admin
13.     password: admin
14.     virtual-host: my_vhost
15.     listener:
16.       simple:
17.         acknowledge-mode: manual
18. mybatis-plus:
19.   typeAliasesPackage: com.xfc.consumer.entities
20.   mapper-locations: classpath:mapper/*.xml

复制代码

3.4.2. RabbitMQ配置类

1. @Configuration
2. public class RabbitMQConfig {
3.     /**
4.      * 死信队列
5.      * @return
6.      */
7.     @Bean
8.     public FanoutExchange deadExchange() {
9.         return new FanoutExchange("dead_xfc_fanout_exchange", true, false);
10.     }
12.     @Bean
13.     public Queue deadXfcQueue() {
14.         return new Queue("dead.xfc.queue", true);
15.     }
16.     @Bean
17.     public Binding bindDeadXfc() {
18.         return BindingBuilder.bind(deadXfcQueue()).to(deadExchange());
19.     }
21.     /**
22.      * 队列
23.      * @return
24.      */
25.     @Bean
26.     public FanoutExchange fanoutExchange() {
27.         return new FanoutExchange("xfc_fanout_exchange", true, false);
28.     }
30.     @Bean
31.     public Queue xfcQueue() {
32.         Map<String, Object> args = new HashMap<>();
33.         args.put("x-dead-letter-exchange", "dead_xfc_fanout_exchange");
34.         return new Queue("xfc.queue", true, false, false, args);
35.     }
37.     @Bean
38.     public Binding bindXfc() {
39.         return BindingBuilder.bind(xfcQueue()).to(fanoutExchange());
40.     }
41. }

复制代码

上述代码为RabbitMQ配置类，用于在项目初始化时天生相应的交换机和队列。 
3.4.3. 队列消耗

1. @Service
2. @Slf4j
3. public class XfcMqConsumer {
4.     @RabbitListener(queues = {"xfc.queue"})
5.     public void messageconsumer(String message, Channel channel,
6.                                 CorrelationData correlationData,
7.                                 @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
8.         MessageConfirmation messageConfirmation=null;
9.         try {
10.             log.info("收到MQ的消息是: " + message );
11.             messageConfirmation= JsonUtil.string2Obj(message, MessageConfirmation.class);
12.             /**
13.              * 编写业务逻辑
14.              */
15.             
16.         } catch (Exception e) {
17.             e.printStackTrace();
18.             log.error("消息投放到死信队列"+e.getMessage(),e);
19.             channel.basicNack(tag,false,false);// 死信队列
20.         }
21.     }
22. }

复制代码

3.4.4. 死信队列消耗

1. @Service
2. @Slf4j
3. public class DeadMqConsumer {
4.     @Autowired
5.     MessageErrorMapper messageErrorMapper;
6.     @RabbitListener(queues = {"dead.xfc.queue"})
7.     public void messageconsumer(String message, Channel channel,
8.                                 CorrelationData correlationData,
9.                                 @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
10.         MessageConfirmation messageConfirmation=null;
11.         try {
12.             log.info("收到MQ的消息是: " + message );
13.             messageConfirmation= JsonUtil.string2Obj(message, MessageConfirmation.class);
14.             /**
15.              * 编写业务逻辑
16.              */
17.         } catch (Exception e) {
18.             e.printStackTrace();
19.             /**
20.              * 写入message_error
21.              */
22.             messageErrorMapper.insert(new MessageError(messageConfirmation.getId(),e.getMessage(),new Date()));
23.             channel.basicNack(tag,false,false);// 死信队列
24.         }
25.     }
26. }

复制代码

3.5 效果测试

以上代码编写完成后需要进行架构效果测试，其步骤如下：
1. 消息投递测试

上图调用了消息投递接口。

在消息确认表中，新增了一条消息且status=1，代表该条消息已投递乐成。
2. 消耗者正常消耗测试

3. 消耗异常测试

上图可看出消息消耗异常投入到了死信队列。

在死信队列中依然消耗失败。

消耗失败后乐成写入了日记表。
4. 结语

本文讲解了RabbitMQ应用场景以及在异步处理场景中如何搭建稳定的RabbitMQ架构体系，逐步具体的给出了生产者及消耗者端代码并在文章最后对架构效果进行了测试，感兴趣的同学可根据代码进行实操，有疑问和其他见解也可在评论区留言，我看到都会回复。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。