RabbitMQ 消息处理问题全解

在使用 RabbitMQ 进行消息队列通信时，可能会遇到消息丢失、乱序、重复消费等问题。这些问题如果不加以妥善处理，可能会导致系统出现数据不一致、业务逻辑错误等严重后果。本文将详细探讨 RabbitMQ 中这些问题的产生原因以及解决方案，并提供丰富的示例帮助读者更好地理解和应用。

 

一、RabbitMQ 简介

 

RabbitMQ 是一个实现了高级消息队列协议（AMQP）的开源消息代理软件。它提供了可靠的、灵活的消息传递机制，广泛应用于分布式系统中，用于解耦系统组件、异步处理任务、提高系统的可扩展性和可靠性。

 

二、消息丢失问题

 

（一）消息丢失的原因

 

  1. 生产者未确认消息发送
    • 当生产者发送消息后，如果没有等待 RabbitMQ 服务器的确认，就不能确定消息是否已经成功发送到了队列中。例如，在网络故障或者 RabbitMQ 服务器出现问题的情况下，消息可能会丢失。
  2. 队列或交换机未持久化
    • 如果队列或交换机没有设置为持久化，那么在 RabbitMQ 服务器重启或者崩溃时，存储在其中的消息可能会丢失。
  3. 消费者未确认消息接收
    • 当消费者接收到消息后，如果没有向 RabbitMQ 服务器发送确认消息，那么在消费者出现故障或者崩溃时，消息可能会被重新投递，导致重复消费的问题。如果消息在重新投递的过程中一直没有被确认，最终可能会被丢弃，导致消息丢失。

 

（二）解决方案

 

  1. 生产者确认机制
    • RabbitMQ 提供了生产者确认机制，允许生产者在发送消息后等待 RabbitMQ 服务器的确认。如果消息成功发送到了队列中，RabbitMQ 服务器会向生产者发送一个确认消息。如果消息发送失败，RabbitMQ 服务器会向生产者发送一个否定确认消息，生产者可以根据这个消息进行重试或者其他处理。
    • 以下是一个使用 Java 语言实现生产者确认机制的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConfirmCallback;public class ProducerWithConfirm {private static final String QUEUE_NAME = "test_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 开启生产者确认机制channel.confirmSelect();ConfirmCallback confirmCallback = (deliveryTag, multiple) -> System.out.println("消息确认：deliveryTag = " + deliveryTag);channel.addConfirmListener(confirmCallback);String message = "Hello, RabbitMQ!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println("消息发送成功：" + message);// 等待确认消息while (!channel.waitForConfirms()) {}}}
}

 

  1. 队列和交换机持久化
    • 在创建队列和交换机时，可以将其设置为持久化，这样即使 RabbitMQ 服务器重启或者崩溃，存储在其中的消息也不会丢失。
    • 以下是一个使用 Java 语言创建持久化队列和交换机的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class CreatePersistentQueueAndExchange {private static final String EXCHANGE_NAME = "test_exchange";private static final String QUEUE_NAME = "test_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 创建持久化交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);// 创建持久化队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 将队列绑定到交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routing_key");}}
}

 

  1. 消费者确认机制
    • 消费者在接收到消息后，应该向 RabbitMQ 服务器发送确认消息，告诉服务器消息已经被成功处理。如果消费者在处理消息的过程中出现故障或者崩溃，RabbitMQ 服务器会将消息重新投递到其他消费者或者等待消费者恢复后重新投递。
    • 以下是一个使用 Java 语言实现消费者确认机制的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class ConsumerWithAck {private static final String QUEUE_NAME = "test_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("接收到消息：" + message);// 发送确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);};// 自动确认关闭channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}}
}

 

三、消息乱序问题

 

（一）消息乱序的原因

 

  1. 多个消费者并行处理消息
    • 如果有多个消费者同时从一个队列中接收消息并进行处理，那么由于消费者的处理速度不同，可能会导致消息的处理顺序与发送顺序不一致。例如，先发送的消息可能会被处理速度较慢的消费者处理，而后发送的消息可能会被处理速度较快的消费者处理，从而导致消息乱序。
  2. 消息重试机制
    • 如果消息在处理过程中出现错误，RabbitMQ 会根据重试机制将消息重新投递到队列中。在重新投递的过程中，消息可能会被其他消费者处理，从而导致消息的处理顺序与发送顺序不一致。

 

（二）解决方案

 

  1. 单个消费者处理消息
    • 为了避免消息乱序问题，可以使用单个消费者来处理消息。这样可以保证消息按照发送顺序依次被处理。但是，这种方法可能会导致处理速度变慢，特别是在处理大量消息时。
  2. 消息顺序标记
    • 在发送消息时，可以为每个消息添加一个顺序标记，例如消息的发送时间戳或者自增的序列号。消费者在接收到消息后，可以根据这个顺序标记来判断消息的顺序，并按照顺序进行处理。如果发现消息乱序，可以将其缓存起来，等待前面的消息处理完成后再进行处理。
    • 以下是一个使用 Java 语言为消息添加顺序标记并进行处理的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.util.ArrayList;
import java.util.List;public class ConsumerWithOrder {private static final String QUEUE_NAME = "test_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback deliverCallback = (consumerTag, delivery) -> {long sequenceNumber = Long.parseLong(new String(delivery.getProperties().getHeaders().get("sequence_number")));System.out.println("接收到消息：sequenceNumber = " + sequenceNumber);List<Long> receivedSequenceNumbers = new ArrayList<>();receivedSequenceNumbers.add(sequenceNumber);while (true) {boolean allInOrder = true;for (int i = 0; i < receivedSequenceNumbers.size() - 1; i++) {if (receivedSequenceNumbers.get(i + 1)!= receivedSequenceNumbers.get(i) + 1) {allInOrder = false;break;}}if (allInOrder) {for (Long receivedSequenceNumber : receivedSequenceNumbers) {System.out.println("处理消息：sequenceNumber = " + receivedSequenceNumber);}receivedSequenceNumbers.clear();} else {break;}}// 发送确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);};// 自动确认关闭channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}}
}

 

四、消息重复消费问题

 

（一）消息重复消费的原因

 

  1. 消费者确认机制问题
    • 如果消费者在处理消息后没有正确地向 RabbitMQ 服务器发送确认消息，那么 RabbitMQ 服务器会认为消息没有被处理，从而将消息重新投递到队列中。如果消费者在重新接收到消息后再次处理，就会导致消息重复消费。
  2. 网络故障或消费者崩溃
    • 在网络故障或者消费者崩溃的情况下，RabbitMQ 服务器可能会将消息重新投递到其他消费者或者等待消费者恢复后重新投递。如果消费者在恢复后没有正确地处理重复的消息，就会导致消息重复消费。

 

（二）解决方案

 

  1. 消息幂等处理
    • 幂等性是指一个操作无论执行多少次，其结果都是相同的。在处理消息时，可以通过实现消息的幂等性来避免重复消费的问题。例如，如果消息是一个数据库插入操作，可以在数据库中设置唯一约束，确保相同的消息不会被重复插入。如果消息是一个文件写入操作，可以在写入文件之前先检查文件是否已经存在相同的内容，如果存在则跳过写入操作。
    • 以下是一个使用 Java 语言实现消息幂等处理的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.util.HashMap;
import java.util.Map;public class ConsumerWithIdempotence {private static final String QUEUE_NAME = "test_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("接收到消息：" + message);// 检查消息是否已经处理过Map<String, Boolean> processedMessages = new HashMap<>();if (processedMessages.containsKey(message)) {System.out.println("消息已经处理过，跳过：" + message);} else {// 处理消息System.out.println("处理消息：" + message);processedMessages.put(message, true);}// 发送确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);};// 自动确认关闭channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}}
}

 

  1. 分布式锁
    • 在处理消息时，可以使用分布式锁来确保同一时间只有一个消费者能够处理特定的消息。如果多个消费者同时接收到相同的消息，只有一个消费者能够获取到分布式锁并处理消息，其他消费者会等待锁释放后再尝试获取锁。这样可以避免消息重复消费的问题。
    • 以下是一个使用 Java 语言实现分布式锁的示例：

  

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import redis.clients.jedis.Jedis;public class ConsumerWithDistributedLock {private static final String QUEUE_NAME = "test_queue";private static final String LOCK_KEY_PREFIX = "message_lock:";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("接收到消息：" + message);// 获取分布式锁Jedis jedis = new Jedis("localhost", 6379);String lockKey = LOCK_KEY_PREFIX + message;boolean locked = false;try {while (!locked) {locked = jedis.setnx(lockKey, "locked") == 1;if (!locked) {Thread.sleep(100);}}// 处理消息System.out.println("处理消息：" + message);// 释放分布式锁jedis.del(lockKey);} finally {jedis.close();}// 发送确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);};// 自动确认关闭channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}}
}

 

五、总结

 

在使用 RabbitMQ 进行消息队列通信时，消息丢失、乱序、重复消费等问题是需要重点关注和解决的。通过合理地使用生产者确认机制、队列和交换机持久化、消费者确认机制、消息顺序标记、消息幂等处理和分布式锁等方法，可以有效地避免这些问题的发生，提高系统的可靠性和稳定性。同时，在实际应用中，还需要根据具体的业务需求和系统架构进行适当的调整和优化，以确保消息队列的高效运行。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/56384.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

深度学习：Sigmoid函数详解

深度学习：Sigmoid函数详解

Sigmoid函数详解 Sigmoid函数是一个广泛应用于神经网络中的激活函数&#xff0c;尤其是在早期的神经网络模型中。其数学表达式如下&#xff1a; [ σ ( x ) 1 1 e − x \sigma(x) \frac{1}{1 e^{-x}} σ(x)1e−x1​ ] 主要特点 输出范围&#xff1a; Sigmoid函数的输出…
阅读更多...
Unity之XR Interaction Toolkit 射线拖拽3DUI

Unity之XR Interaction Toolkit 射线拖拽3DUI

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、想实现的功能二、实现原理1.UI挂在XRGrabInteractable、刚体、BoxCollder2.修改刚体属性3.加BoxCollder 总结 前言 VR项目里正常情况有放置两种3DUI的方式…
阅读更多...
如何在 HarmonyOS NEXT 中使用 @Builder 装饰器优化 UI 组件的复用？

如何在 HarmonyOS NEXT 中使用 @Builder 装饰器优化 UI 组件的复用？

摘要 在鸿蒙 NEXT 开发中&#xff0c;Builder 装饰器是一种轻量级的 UI 元素复用机制&#xff0c;它允许开发者将重复使用的 UI 元素抽象成一个方法&#xff0c;并在 build() 方法中多次调用&#xff0c;以实现 UI 结构的复用。以下是如何使用 Builder 装饰器来优化 UI 组件复…
阅读更多...
Ovis: 多模态大语言模型的结构化嵌入对齐

Ovis: 多模态大语言模型的结构化嵌入对齐

论文题目&#xff1a;Ovis: Structural Embedding Alignment for Multimodal Large Language Model 论文地址&#xff1a;https://arxiv.org/pdf/2405.20797 github地址&#xff1a;https://github.com/AIDC-AI/Ovis/?tabreadme-ov-file 今天&#xff0c;我将分享一项重要的研…
阅读更多...
关于使用 C# 处理水位数据多种格式的统一转换

关于使用 C# 处理水位数据多种格式的统一转换

关于使用 C# 处理水位数据多种格式的统一转换 1、前言2、水位数据的多种格式3、水位数据多种格式的统一转换程序展示4、水位数据多种格式的统一转换 C# 代码4.1、声明引用命名空间4.2、多种格式的统一转换 C# 代码4.3、多种格式的统一转换 C# 代码&#xff0c;文件输出保存 1、…
阅读更多...
Django进一步掌握（10月22日）

Django进一步掌握（10月22日）

一、请求响应对象 请求对象request 响应对象HttpResponse 二、HttpResponse常用属性 status设置HTTP响应状态码 status_code查询HTTP响应状态码 content_type设置响应的类型 write()写入响应内容 三、重定向 1、实现URl访问的重定向 &#xff08;1&#xff09;使用Ht…
阅读更多...
【尊享面试100题】数组/字符串

【尊享面试100题】数组/字符串

多看优秀的代码 1.数组列表中的最大距离2.字符串的左右移3.相隔为1的编辑距离4.形成字符串的最短路径5.连接二进制表示可形成的最大数值 1.数组列表中的最大距离 给定 m 个数组&#xff0c;每个数组都已经按照升序排好序了。 现在你需要从两个不同的数组中选择两个整数&#…
阅读更多...
微知-Lecroy力科的PCIe协议分析仪型号命名规则(PCIe代，金手指lanes数量)

微知-Lecroy力科的PCIe协议分析仪型号命名规则(PCIe代，金手指lanes数量)

文章目录 要点主要型号命名规则各代主要产品图片Summit M616 协议分析仪/训练器Summit T516 分析仪Summit T416 分析仪Summit T3-16分析仪Summit T28 分析仪 综述 要点 LeCroy(力科)成立于1964年&#xff0c;是一家专业生产示波器厂家。在美国纽约。一直把重点放在研制改善生产…
阅读更多...
Hallo2 长视频和高分辨率的音频驱动的肖像图像动画 （数字人技术）

Hallo2 长视频和高分辨率的音频驱动的肖像图像动画 （数字人技术）

HALLO2: LONG-DURATION AND HIGH-RESOLUTION AUDIO-DRIVEN PORTRAIT IMAGE ANIMATION 论文&#xff1a;https://arxiv.org/abs/2410.07718 代码&#xff1a;https://github.com/fudan-generative-vision/hallo2 模型&#xff1a;https://huggingface.co/fudan-generative-ai/h…
阅读更多...
TikTok营销实用技巧与数据分析工具：视频洞察

TikTok营销实用技巧与数据分析工具：视频洞察

TikTok凭借其独特的机制和庞大的流量&#xff0c;成为了众多品牌和卖家对产品进行宣传推广的必要平台之一。要在TikTok上优化营销效果、提升推广效率&#xff0c;可以使用平台提供的重要工具——视频洞察&#xff08;Video Insights&#xff09;。 一、视频洞察功能与技巧 视频…
阅读更多...
React Native 项目使用Expo模拟器运行iOS和Android

React Native 项目使用Expo模拟器运行iOS和Android

iOS没有连接设备&#xff1a; 确保你已经用 USB 线将你的 iOS 设备连接到了你的 Mac。 设备未信任&#xff1a; 如果你的设备是第一次连接到 Mac&#xff0c;可能需要在设备上信任这台计算机。通常&#xff0c;当你连接设备时&#xff0c;设备上会弹出一个对话框&#xff0c;…
阅读更多...
线性回归（一）

线性回归（一）

线性回归 1.基本术语 ①特征&#xff1a;预测所依据的自变量称为特征或协变量 ②标签&#xff1a;试图预测的目标称为标签或目标 2.举个栗子 线性假设是指目标&#xff08;房屋价格&#xff09;可以表示为特征&#xff08;面积和房龄&#xff09;的加权和&#xff0c;如下面…
阅读更多...
YOLOv11入门到入土使用教程(含结构图)

YOLOv11入门到入土使用教程(含结构图)

一、简介 YOLOv11是Ultralytics公司在之前的YOLO版本上推出的最新一代实时目标检测器&#xff0c;支持目标检测、追踪、实力分割、图像分类和姿态估计等任务。官方代码&#xff1a;ultralytics/ultralytics&#xff1a;ultralytics YOLO11 &#x1f680; (github.com)https://g…
阅读更多...
解决跨域问题

解决跨域问题

跨域是浏览器受同源策略的限制&#xff0c;同源策略是浏览器为确保资源安全&#xff0c;而遵循的一种策略&#xff0c;该策略对访问资源进行了一些限制&#xff08;如发送 ajax 请求&#xff0c;操作 dom&#xff0c;读取 cookie&#xff09;。 最常见的影响就是发送 ajax 请求…
阅读更多...
【微知】如何通过命令行在非串口界面触发sysrq的help信息？（echo h ＞ /proc/sysrq-trigger）

【微知】如何通过命令行在非串口界面触发sysrq的help信息？（echo h ＞ /proc/sysrq-trigger）

背景 在服务器上&#xff0c;触发sysrq通常需要在串口执行sysrq热键&#xff0c;比如 ~相关的操作 如何通过在ssh界面触发sysrq触发一些操作&#xff1f; 命令 通过sysrq指定的/proc接口文件进行操作 echo h > /proc/sysrq-trigger dmesg #产看输出的帮助信息然后根据打…
阅读更多...
Junit + Mockito保姆级集成测试实践

Junit + Mockito保姆级集成测试实践

一、做好单测&#xff0c;慢即是快 对于单元测试的看法&#xff0c;业界同仁理解多有不同&#xff0c;尤其是在业务变化快速的互联网行业&#xff0c;通常的问题主要有&#xff0c;必须要做吗&#xff1f;做到多少合适&#xff1f;现在没做不也挺好的吗&#xff1f;甚至一些大…
阅读更多...
MYSQL-SQL-01-DDL（Data Definition Language，数据定义语言）

MYSQL-SQL-01-DDL（Data Definition Language，数据定义语言）

DDL&#xff08;数据定义语言&#xff09; DDL&#xff08;Data Definition Language&#xff09;&#xff0c;数据定义语言&#xff0c;用来定义数据库对象(数据库&#xff0c;表&#xff0c;字段) 。 一、数据库操作 1、 查询mysql数据库管理系统的所有数据库 语法&#…
阅读更多...
django（3）jinja2模版的使用

django（3）jinja2模版的使用

启动模版 安装jinja2 pip install jinja2 配置setting TEMPLATES中添加配置 {BACKEND: django.template.backends.jinja2.Jinja2,DIRS: [os.path.join(BASE_DIR,jinja2)], #模版在项目中的所在位置} template中各项的含义 这个配置项中模版自上而下加载&#xff0c;重名…
阅读更多...
Spring Boot框架的电影评论系统设计与实现

Spring Boot框架的电影评论系统设计与实现

3系统分析 3.1可行性分析 通过对本电影评论网站实行的目的初步调查和分析&#xff0c;提出可行性方案并对其一一进行论证。我们在这里主要从技术可行性、经济可行性、操作可行性等方面进行分析。 3.1.1技术可行性 本电影评论网站采用SSM框架&#xff0c;JAVA作为开发语言&#…
阅读更多...
基于单片机的智能小区门禁系统设计（论文+源码）

基于单片机的智能小区门禁系统设计（论文+源码）

1总体架构 智能小区门禁系统以STM32单片机和WiFi技术为核心&#xff0c;STM32单片机作为主控单元&#xff0c;通过WiFi模块实现与手机APP的连接&#xff0c;构建整个门禁系统。系统硬件包括RFID模块、指纹识别模块、显示屏、按键以及继电器。通过RFID绑定IC卡、APP面部识别、指…
阅读更多...
最新文章

Copyright @ 2022~2023