RabbitMQ高级特性详解

前言

RabbitMQ是一款广泛使用的开源消息队列软件,它基于AMQP(Advanced Message Queuing Protocol)标准实现。本文将带你深入了解RabbitMQ的一些高级特性,包括消息确认、死信队列、延迟队列、事务处理以及消息分发策略等,并通过示例代码展示如何在实际项目中应用这些特性。


目录

1. 消息确认机制

1.1 什么是消息确认?

自动确认 vs 手动确认

1.2 使用场景

2. 死信队列

2.1 死信的概念与来源

2.2 应用场景

3. 延迟队列

3.1 延迟队列简介

3.2 实现方法

示例代码

3.3 场景应用

4. 事务处理

4.1 事务的基本概念

4.2 配置与使用

4.3 生产者示例

5. 消息分发

5.1 分发机制

5.2 限流与负载均衡

5.3 示例代码

总结


1. 消息确认机制

1.1 什么是消息确认?

消息确认是确保消息从队列可靠地到达消费者的关键机制。生产者发送消息后,消息可能会被成功处理,也可能因为异常而未被正确处理。为了保证消息的可靠性,RabbitMQ引入了消息确认机制。

自动确认 vs 手动确认
  • 自动确认:当autoAck设置为true时,RabbitMQ会自动认为消息一旦发送出去就被消费端接收到了,并立刻从队列中删除这条消息。这种方式适合对消息可靠性要求不高的场景。
  • 手动确认:当autoAck设置为false时,RabbitMQ会在接收到消费者的显式确认信号后才会移除消息。这种模式适用于需要高可靠性的场景。
// 自动确认示例
channel.basicConsume("queueName", true, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息: " + new String(body));}
});// 手动确认示例
channel.basicConsume("queueName", false, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try {// 处理消息System.out.println("接收到消息: " + new String(body));// 确认消息channel.basicAck(envelope.getDeliveryTag(), false);} catch (Exception e) {// 异常处理,拒绝签收channel.basicNack(envelope.getDeliveryTag(), false, true); // requeue设为true表示重新入队}}
});

1.2 使用场景

使用自动确认可以简化开发流程,但可能会导致消息丢失。手动确认虽然增加了额外的确认步骤,但是能显著提高消息传递的可靠性,特别适用于金融交易、订单处理等关键业务场景。


2. 死信队列

2.1 死信的概念与来源

死信是指那些无法被正常消费的消息。它们可能由于以下几种情况产生:

  • 消息过期:消息在队列中的存活时间超过了设定的时间阈值(TTL)。
  • 消息被拒绝:消费者在处理过程中遇到错误并拒绝了该消息。
  • 队列满载:当队列达到最大长度限制时,新来的消息会被视为死信。

2.2 应用场景

  • 消息重试:将未能处理的消息重新发送到原始队列或另一个队列进行尝试。
  • 消息丢弃:直接丢弃那些不可处理的消息以避免占用资源。
  • 日志收集:将死信作为日志记录下来,以便后续分析问题所在。
@Bean
public Queue normalQueue() {Map<String, Object> arguments = new HashMap<>();arguments.put("x-dead-letter-exchange", "dlxExchange");arguments.put("x-dead-letter-routing-key", "dlxRoutingKey");return QueueBuilder.durable("normalQueue").withArguments(arguments).build();
}@RabbitListener(queues = "dlxQueue")
public void listenerDLXQueue(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.printf("死信队列接收到消息: %s, deliveryTag: %d%n", new String(message.getBody()), deliveryTag);
}

3. 延迟队列

3.1 延迟队列简介

延迟队列是一种特殊的队列,其中的消息不是立即被消费,而是等待指定的时间后才可被获取。RabbitMQ本身并不支持延迟队列功能,但可以通过结合TTL(Time to Live)和死信队列来模拟实现。

3.2 实现方法

  • TTL+死信队列:给消息设置生存时间,超时后转至死信队列。
  • 官方插件:使用RabbitMQ提供的延迟消息插件。
示例代码
@Bean
public Queue delayedQueue() {Map<String, Object> args = new HashMap<>();args.put("x-dead-letter-exchange", "dlxExchange");args.put("x-message-ttl", 5000); // 设置消息TTL为5秒return QueueBuilder.durable("delayedQueue").withArguments(args).build();
}@RabbitListener(queues = "dlxQueue")
public void listenDlxQueue(Message message, Channel channel) throws Exception {System.out.printf("%tc 死信队列接收到消息: %s%n", new Date(), new String(message.getBody()));
}

3.3 场景应用

  • 用户注册后发送激活邮件。
  • 订单系统中未支付订单的自动取消。
  • 退款请求处理后的自动退款。

4. 事务处理

4.1 事务的基本概念

RabbitMQ支持事务处理,允许开发者确保消息的发送和接收是原子操作。这意味着要么全部完成,要么全部失败,从而保持数据的一致性。

4.2 配置与使用

配置事务管理器并开启事务支持:

@Configuration
public class TransactionConfig {@Beanpublic RabbitTransactionManager transactionManager(CachingConnectionFactory connectionFactory) {return new RabbitTransactionManager(connectionFactory);}@Beanpublic RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setChannelTransacted(true);return template;}
}

4.3 生产者示例

@Transactional
@RequestMapping("/send")
public String send() {rabbitTemplate.convertAndSend("", "transQueue", "trans test 1...");int a = 5 / 0; // 故意引发异常rabbitTemplate.convertAndSend("", "transQueue", "trans test 2...");return "发送成功";
}

如果启用了事务,上面的代码将会回滚整个操作,确保没有消息被发送出去。


5. 消息分发

5.1 分发机制

当一个队列有多个消费者时,RabbitMQ会根据一定的规则将消息分配给不同的消费者。默认情况下,采用轮询方式分发,即每个消费者轮流获得一条消息。这种方法可能导致某些快速消费者空闲,而慢速消费者积压大量消息的问题。

5.2 限流与负载均衡

  • 限流:通过设置basicQos参数来控制单个消费者同时处理的消息数量,防止过载。
  • 负载均衡:设置prefetchCount=1,使得RabbitMQ每次只向一个消费者发送一条消息,在收到确认之前不会发送新的消息,从而实现更公平的负载分配。
# application.yml
spring:rabbitmq:listener:simple:acknowledge-mode: manualprefetch: 1

5.3 示例代码

@Component
public class QosQueueListener {@RabbitListener(queues = "qosQueue")public void listenQosQueue(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.printf("接收到消息: %s, deliveryTag: %d%n", new String(message.getBody()), deliveryTag);Thread.sleep(100); // 模拟耗时操作channel.basicAck(deliveryTag, false);}
}

总结

  1. 消息确认机制

    • 通过autoAck参数控制消息的确认方式。
    • 自动确认模式适合对消息可靠性要求不高的场景;手动确认模式适用于需要高可靠性的场景。
    • 手动确认允许消费者显式地向RabbitMQ发送确认信号,确保消息被成功处理后才从队列中移除。
  2. 死信队列(Dead Letter Exchange, DLX)

    • 死信是指那些无法被正常消费的消息,可能由于消息过期、被拒绝或队列满载等原因产生。
    • 死信队列可以用来存储这些无法处理的消息,提供重试、丢弃或日志记录等功能。
    • 使用示例展示了如何配置普通队列与DLX绑定,并设置消息TTL和路由键来实现消息转移至DLX。
  3. 延迟队列

    • 延迟队列用于在指定时间之后才将消息传递给消费者。
    • 通过结合TTL与DLX,或者使用官方提供的延迟插件,可以模拟实现延迟队列功能。
    • 这种机制常应用于定时任务、订单超时处理等场景。
  4. 事务处理

    • RabbitMQ支持事务操作,确保消息的发送和接收是原子性的。
    • 通过配置事务管理器并开启事务支持,可以在发生异常时回滚整个事务,保持数据一致性。
    • 示例代码演示了如何在Spring应用中配置并使用事务管理。
  5. 消息分发策略

    • 当多个消费者订阅同一个队列时,RabbitMQ会根据一定的规则分配消息。
    • 默认采用轮询分发,但可以通过设置basicQos参数进行限流,以及调整prefetchCount实现更公平的负载均衡。
    • 适当配置有助于避免某些消费者过载而其他消费者空闲的问题,提高系统的整体吞吐量和稳定性。

以上就是 RabbitMQ 的部分高级特性,有问题可在评论区讨论,感谢阅览!!

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

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

相关文章

HCIE-Datacom题库_01_防火墙【18道题】

一、单选题 1.相比较于路由器、交接机&#xff0c;防火墙转发独有的模块为? 交换网板 MPU LPU SPU 解析&#xff1a; SFU&#xff08;Switch Fabric Unit&#xff09;&#xff1a;交换网板&#xff0c;负责整个系统的数据平面数据平面提供高速无阻塞数据通道&#xff0…

关于SOCKS协议的常见误区有哪些?

代理协议在设备与代理服务器之间的数据交换中起到了关键作用。在这方面&#xff0c;SOCKS代理协议是常见的选择之一&#xff0c;被广泛应用于下载、传输和上传网络数据的场景。然而&#xff0c;关于SOCKS代理协议存在一些常见的误解&#xff0c;让我们来逐一了解。 一、使用SO…

爬虫逆向-js进阶(续写,搭建网站)

1.搭建简单网站1 from flask import Flask,render_template import requests import json app Flask(name)# **location**的温度是**temp**度&#xff0c;天气状况&#xff1a;**desc**app.route(/) # 绑定处理函数 def index_url():location 101010100data get_weather(lo…

Whisper 音视频转写

Whisper 音视频转写 API 接口文档 api.py import os import shutil import socket import torch import whisper from moviepy.editor import VideoFileClip import opencc from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request from fastapi.respons…

Android中的内存泄漏及其检测方式

Android中的内存泄漏及其检测方式 一、Android内存泄漏概述 在Android开发中&#xff0c;内存泄漏是一个常见且严重的问题。内存泄漏指的是在应用程序中&#xff0c;由于某些原因&#xff0c;已经不再使用的对象仍然被引用&#xff0c;导致垃圾回收器&#xff08;Garbage Col…

图书管理新纪元:Spring Boot进销存系统

1系统概述 1.1 研究背景 随着计算机技术的发展以及计算机网络的逐渐普及&#xff0c;互联网成为人们查找信息的重要场所&#xff0c;二十一世纪是信息的时代&#xff0c;所以信息的管理显得特别重要。因此&#xff0c;使用计算机来管理图书进销存管理系统的相关信息成为必然。开…

使用 Elasticsearch Dump 工具进行生产环境到测试环境的数据迁移与备份

es-dump 是 Elasticsearch 的一个实用工具&#xff0c;专门用于从 Elasticsearch 集群中导出或导入数据&#xff0c;支持数据、映射、别名、模板等多种类型的数据操作。它在数据迁移、备份、恢复等场景中非常实用。本文将展示如何使用 es-dump 工具执行生产到测试环境的索引复制…

【学习】word保存图片

word中有想保存的照片 直接右键另存为的话&#xff0c;文件总是不清晰&#xff0c;截屏的话&#xff0c;好像也欠妥。 怎么办? 可以另存为 网页 .html 可以得到&#xff1a; 原图就放到了文件夹里面

Selenium 中定位元素操作集合

Selenium中元素的定位与操作 在 Selenium 中定位单个元素一般通过 find_element 方法来实现&#xff0c;一组数据则是 find_elements &#xff08;定位时应该元素加载完整再 再定位&#xff0c;否则可能会出现找不到对应元素的情况出现。这里可以使用 time.sleep() 来实现等待…

C++简易日志系统:打造高效、线程安全的日志记录工具

目录 引言&#xff1a; 1.日志的基本概念 1.1.什么是日志&#xff1f; 1.2.我们为什么需要日志&#xff1f; 2.自己实现一个简易日志 2.1.日志的等级 2.2日志的格式 2.3.获取时间的方法 2.4.日志的主体实现 参数&#xff1a; 代码解析&#xff1a; 问题&#xff1a…

【ssh 密钥生成】添加后提示无权限

记录下又一个感觉自己是菜菜的事件&#xff1b; 我根据教程在新电脑生成了 ssh 密匙&#xff0c;也在页面添加了ssh密匙&#xff1b; 但是下载代码的时候就提示权限不对&#xff0c;下载不下来&#xff1b; 百度了一堆办法&#xff0c;还是不行&#xff0c;然后大佬来&#…

solid wrok笔记记录

1.平移 crtl鼠标滚轮 2.放大缩小 鼠标滚轮滚动 3.旋转 按住鼠标滚轮 4.正视 右键 选择 5。多选 ctel 框选。。。 6. 选项卡 右上角空白处右键 7.草图&#xff0c;剪多余的线段(T),鼠标含住滑动去除 8.草图的线段掰直 选中线段后&#xff0c;左侧窗口的几…

【Scala入门学习】匹配模式match

1. match匹配 match 语句用在当需要从多个分支中进行选择的场景&#xff0c;类似于java 中的switch 语句。 语法&#xff1a; 变量 match{case "值" > 语句块1 // 语句块后不用加breakcase "值2" > 语句块2case _ > 语句块N // 类似于jav…

5、JavaScript(五)

28.jquery&#xff1a;js库 简化版本的js&#xff0c;封装了现成功能的js代码。 jquery就是一些封装好了的现成的方法&#xff0c;供我们直接使用。 jquery能实现的js都能实现。 在使用 记得先引入jquery&#xff1a;在菜鸟教程上直接用jquery的绝对路径引入&#xff0c;jq…

Gin框架操作指南03:HTML渲染

官方文档地址&#xff08;中文&#xff09;&#xff1a;https://gin-gonic.com/zh-cn/docs/ 注&#xff1a;本教程采用工作区机制&#xff0c;所以一个项目下载了Gin框架&#xff0c;其余项目就无需重复下载&#xff0c;想了解的读者可阅读第一节&#xff1a;Gin操作指南&#…

java游戏网站源码

题目&#xff1a;java游戏网站源码 编号B22A390 主要内容&#xff1a;毕业设计(Javaweb项目|小程序|Mysql|大数据|SSM|SpringBoot|Vue|Jsp|MYSQL等)、学习资料、JAVA源码、技术咨询 文末联系获取 感兴趣可以先收藏起来&#xff0c;以防走丢&#xff0c;有任何选题、文档编…

什么是 BloomFilter

什么是 BloomFilter 布隆过滤器&#xff08;英语&#xff1a;Bloom Filter&#xff09;是 1970 年由布隆提出的。它实际上是一个很长的二进制向量和一系列随机映射函数。主要用于判断一个元素是否在一个集合中。 通常我们会遇到很多要判断一个元素是否在某个集合中的业务场景&a…

Cocos Creator导出obj文件用于后端寻路

Cocos Creator 3.8.0 用这个扩展插件 【杨宗宝】两年前写的网格工具&#xff0c;今天将它开源了。 - Creator 3.x - Cocos中文社区carlosyzy_extensions_mesh: Cocos Creator 3.x mesh插件&#xff0c;负责网格数据的导出。合并&#xff0c;拆封等一系列操作 (gitee.com) 下…

C++ 标准库:功能与应用解析

C++ 标准库:功能与应用解析 引言 C++ 是一种广泛使用的编程语言,以其高性能和灵活性而著称。C++ 标准库(C++ Standard Library)是 C++ 语言的核心组成部分,提供了一系列预定义的类和函数,用于简化编程任务。本文将深入探讨 C++ 标准库的主要功能和应用,帮助读者更好地…

Shell脚本备份文件

需求&#xff1a;原文件备份&#xff0c;新文件覆盖掉源文件 vue项目打包自动化部署使用 假设已经将打包后得文件复制到了需要覆盖得目录得同级下 #!/bin/bashdst_folder"/home/compose/nginx/html"# 创建备份文件夹 backup_folder"$dst_folder/backup"# …