02_RabbitMQ消息丢失解决方案及死信队列

一、数据丢失

第一种:生产者弄丢了数据。生产者将数据发送到 RabbitMQ 的时候,可能数据就在半路给搞丢了,因为网络问题,都有可能。

第二种:RabbitMQ 弄丢了数据。MQ还没有持久化自己挂了。

第三种:消费端弄丢了数据。刚消费到,还没处理,结果进程挂了,比如重启了。

解决方案

消费者从队列中获取到数据,应答成功之后,队列认为消费者对消息处理完成,从队列中删除消息。

生产者

mq告知生产者,消息接收成功,没有成功,生产者可以继续重新发送消息

RabbitMQ有两种方式来解决这个问题:

  • 通过AMQP提供的事务机制实现(同步,不推荐);

  • 使用发送者确认模式实现(confirm,异步);

事务使用

事务的实现主要是对信道(Channel)的设置,主要的方法有三个:

  1. channel.txSelect()声明启动事务模式;

  2. channel.txCommit()提交事务;

  3. channel.txRollback()回滚事务;

public class Product {
​
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnUtils.getConn();
        Channel channel = conn.createChannel();
​
        String queueName = "my-que";
        // queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)
        channel.queueDeclare(queueName,false,false,false,null);
​
        String message = "hello rabbitmq";
​
        try{
            // 开启事务
            channel.txSelect();
            // 发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
            // 提交事务
            channel.txCommit();
        }catch (Exception e){
            e.printStackTrace();
            // 回滚事务
            channel.txRollback();
        }
        
        channel.close();
        conn.close();
    }
}
Confirm发送方确认模式

Confirm发送方确认模式使用和事务类似,也是通过设置Channel进行发送方确认的。

Confirm的三种实现方式:

  • channel.waitForConfirms()普通发送方确认模式,单条数据;

  • channel.waitForConfirmsOrDie()批量确认模式,多条数据;

  • channel.addConfirmListener()异步监听发送方确认模式;

public class Product {
​
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnUtils.getConn();
        // 创建信道:处理消息
        Channel channel = conn.createChannel();
​
        // 创建队列:存储消息
        String queueName = "my-que";
        // queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)
        channel.queueDeclare(queueName, false, false, false, null);
​
        // 定义消息
        String message = "hello rabbitmq";
​
        // 开启发送方确认模式
        channel.confirmSelect();
        // basicPublish(交换器,序列名,参数信息,消息)
        channel.basicPublish("", queueName, null, message.getBytes());
​
        if(channel.waitForConfirms()){
            System.out.println("发送成功");
        }
​
        channel.close();
        conn.close();
    }
}
// 开启发送方确认模式
channel.confirmSelect();
​
for (int i = 0; i < 10; i++) {
    channel.basicPublish("", queueName, null, message.getBytes());
}
​
// 直到所有信息都发布,只要有一个未确认就会IOException
channel.waitForConfirmsOrDie(); 
System.out.println("全部执行完成");
// 开启发送方确认模式
channel.confirmSelect();
​
for (int i = 0; i < 10; i++) {
    channel.basicPublish("", queueName, null, message.getBytes());
}
​
//异步监听确认和未确认的消息
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("未确认消息,标识:" + deliveryTag);
    }
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println(String.format("已确认消息,标识:%d,多个消息:%b", deliveryTag, multiple));
    }
});
​
System.out.println("执行操作...");

MQ

rabbitmq做持久化,持久化主要分为:

  • 交换器的持久化

    // 参数1 exchange :交换器名
    // 参数2 type :交换器类型
    // 参数3 durable :是否持久化
    channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
  • 队列持久化

    // 参数1 queue :队列名
    // 参数2 durable :是否持久化
    // 参数3 exclusive :仅创建者可以使用的私有队列,断开后自动删除
    // 参数4 autoDelete : 当所有消费客户端连接断开后,是否自动删除队列
    // 参数5 arguments
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
  • 消息持久化

    // 参数1 exchange :交换器
    // 参数2 routingKey : 路由键
    // 参数3 props : 消息的其他参数,其中 MessageProperties.PERSISTENT_TEXT_PLAIN 表示持久化
    // 参数4 body : 消息体
    channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

消费者

关闭自动应答,改为手动应答,处理完成,MQ中再做删除

        // 消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("deliveryTag:" + envelope.getDeliveryTag());
                Thread.sleep(1000);
                System.out.println("获取到的消息内容:" + new String(body));
                // 手动应答
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
​
        // false:关闭自动应答
        channel.basicConsume(queueName, false, consumer);

二、死信队列

  • 消息过期会自动删除

  • 消费者立刻接收消息进行处理

消费者立刻接收到数据处理,需求队列得数据不立刻被消费者处理,而是希望等待一段时间才被处理

死信队列:是延迟队列得一种实现方式,不是一种特殊得队列,死信交换机+消息有效期

  • 交换机的类型:fanout,direct,topic,header

  • 队列:不区分类型,也是一个普通得队列

  • 死信交换机:将一个交换机设置为某一个队列的死信交换机,当前队列过期的消息会发送给交换机

  • 消息:有效期

消息传递顺序

生产者发布消息 --> 交换机(转发消息) --> 队列(接收存储消息) --> 消费者(接收处理消息)

队列转发消息:

  1. 正常消息:转给消费者处理;

  2. 过期消息:默认删除,也可以通过配置(死信交换机)转发给交换机;

交换机的消息来源:

  1. 生产者发布的(普通得生产者)

  2. 其他队列到期删除的消息自动进入队列(队列需要绑定交换机,队列发布)

队列指的内容:

  • 消息来源都是从交换机发送得

死信队列:交换机B作为队列A的死信交换机,接收队列A中被删除的消息,实现队列A中的消息延迟处理的效果

死信队列DLX(dead-letter-exchange),利用DLX,当消息在一个队列中变成死信 (dead message) 之后,它能被重新投递到另一个exchange上,这个exchange就是DLX,似乎叫死信交换机更加贴切,当死信投递到这个exchange后,我们也可以用一个queue来绑定该exchange,该exchange就可以根据路由规则把这个死信路由到这个queue了,同样可以创建消费端去进行消费,以便对死信进行相应的处理。

延时队列:实际是不存在直接可用的延时队列,可通过死信消息和死信队列来实现延时队列的功能。RabbitMQ中的所有队列,消息存进来,立刻分发给对应的消费者进行处理。

死信交换机: DLX 全称(Dead-Letter-Exchange)。其实它是个普通的交换机,但它是设置在队列上某个参数的值对应的交换机。

死信队列:如果某个队列上存在参数:x-dead-letter-exchange, 当这个队列里的消息变成死信消息(dead message)后会被重新Pushlish到 x-dead-letter-exchange 所对应参数值的交换机上,跟这个交换机所绑定的队列就是死信队列。

为什么使用延迟队列

订单五分钟内支付:未支付订单超时

以前:1.用户做支付时,判断是否超时,修改订单状态;(不调用)

​ 2.定时任务,每一秒刷新一次;(频繁刷新)

延时队列的应用场景很多,在我的项目开发中也涉及到很多,例如:订单五分钟未支付自动取消、订单准备超时30分钟推送提醒给门店、订单完成后两小时推送评价邀请给用户等等,这些间隔指定时间后的操作都可以使用延时队列。

消息变成死信的情况

死信消息

  • 消息被拒绝(basic.reject / basic.nack),并且requeue = false(不重回队列)。

  • 消息TTL过期。

  • 队列达到最大长度。

过期消息:RabbitMq 有两种设置消息过期的方式:

  1. 创建队列时通过 x-message-ttl 参数指定该队列消息的过期时间,这种队列里的消息过期时间全部相同。

  2. 生产者Pushlish消息时,通过设置消息的 expiration 参数指定过期时间,每个消息的过期时间都不一样。

  如果两者同时使用,过期时间按照小的一方为准,两种方式设置的时间都是 毫秒。

Queue参数信息

  • name: 队列的名称;

  • actualName: 队列的真实名称,默认用name参数,如果name为空,则根据规则生成一个;

  • durable: 是否持久化;

  • exclusive: 是否独享、排外的;

  • autoDelete: 是否自动删除;

  • arguments:队列的其他属性参数,有如下可选项,可参看图2的arguments:

    • x-message-ttl:消息的过期时间,单位:毫秒;

    • x-expires:队列过期时间,队列在多长时间未被访问将被删除,单位:毫秒;

    • x-max-length:队列最大长度,超过该最大值,则将从队列头部开始删除消息;

    • x-max-length-bytes:队列消息内容占用最大空间,受限于内存大小,超过该阈值则从队列头部开始删除消息;

    • x-overflow:设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。有效值是drop-head、reject-publish或reject-publish-dlx。仲裁队列类型仅支持drop-head;

    • x-dead-letter-exchange:死信交换器名称,过期或被删除(因队列长度超长或因空间超出阈值)的消息可指定发送到该交换器中;

    • x-dead-letter-routing-key:死信消息路由键,在消息发送到死信交换器时会使用该路由键,如果不设置,则使用消息的原来的路由键值

    • x-single-active-consumer:表示队列是否是单一活动消费者,true时,注册的消费组内只有一个消费者消费消息,其他被忽略,false时消息循环分发给所有消费者(默认false)

    • x-max-priority:队列要支持的最大优先级数;如果未设置,队列将不支持消息优先级;

    • x-queue-mode(Lazy mode):将队列设置为延迟模式,在磁盘上保留尽可能多的消息,以减少RAM的使用;如果未设置,队列将保留内存缓存以尽可能快地传递消息;

    • x-queue-master-locator:在集群模式下设置镜像队列的主节点信息。

交换机参数信息

类型

  • FanoutExchange :扇形 交换机

  • DirectExchange :直连交换机 routing模式用的交换机

  • TopicExchange :主题模式交换机

参数信息

  • name:交换机名称

  • durable:是否把交换机持久化到磁盘上

  • autoDelete :是否自动删除交换机 用法有点类似上面的队列中的autoDelete 只有所有队列都和交换机接触订阅,说白了就是所有绑定到交换机上的队列不再需要改交换机,他就该死了,有点残酷

  • artuments :额外参数

实现

生产者

发布消息,设置消息过期时间

public class Producer {
​
    // default exchange
    private static String exchange = "test_dlx_exchange";
    // default exchange 的路由规则: routingKey(test) 将匹配同名的 queue(test)
    private static String routingKey = "dlx.abc";
​
    public static void main(String[] args) throws IOException, TimeoutException {
​
        // 1 创建Connection
        Connection connection = ConnUtils.getConn();
​
        // 2 创建Channel
        Channel channel = connection.createChannel();
​
        // 3 发送消息
        for (int i = 0; i < 5; i++) {
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    // deliveryMode:1(nopersistent)非持久化,2(persistent)持久化
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    // expiration:消息的失效时间
                    .expiration("50000")
                    .build();
            String msg = "RabbitMQ: dlx message" + i;
            channel.basicPublish(exchange , routingKey , properties , msg.getBytes());
        }
​
        // 5 关闭连接
        channel.close();
        connection.close();
    }
}

正常消费者

绑定死信队列,当消息变为死信消息时,会自动存入私信队列

public class Consumer {
​
    public static void main(String[] args) throws IOException, TimeoutException {
​
        Connection connection = ConnUtils.getConn();
​
        Channel channel = connection.createChannel();
​
        // 定义死信队的Exchange
        String dlxExchange = "dlx.exchange";
        channel.exchangeDeclare(dlxExchange, "topic");
​
        // 死信队列名
        String dlxQueue = "dlx.queue";
        channel.queueDeclare(dlxQueue, true, false, false, null);
​
        // # 表示所有的key都可以路由到s死信队列
        String dlxRoutingKey = "#";
        // 绑定死信队列和exchange
        channel.queueBind(dlxQueue, dlxExchange, dlxRoutingKey, null);
​
        // 定义正常的消费者j监听队列
        // 申明exchange
        String exchangeName = "test_dlx_exchange";
        channel.exchangeDeclare(exchangeName, "topic");
​
        // 申明队列
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信队列,arguments要设置到申明的队列上
        arguments.put("x-dead-letter-exchange", dlxExchange);
        String queueName = "test_dlx_queue";
        channel.queueDeclare(queueName, true, false, false, arguments);
​
        // 队列绑定到exchange
        String routingKey = "dlx.#";
        channel.queueBind(queueName, exchangeName, routingKey);
​
        channel.basicQos(1);
​
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("---消费者-- " + new String(message.getBody(), "UTF-8"));
​
            }
        };
​
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("---消费者--:cancelCallback ");
            }
        };
​
        // callback: 消费者会调函数,处理发送过来的消息。
        // cancelCallback: 消费者取消订阅时的回调方法。
        // 消费消息,autoAck一定要设为false,手工ack
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback);
    }
}

死信消费者

处理死信队列中得信息

// DLXConusmer,监听消费死信队列中的消息
public class DLXConusmer {
​
    public static void main(String[] args) throws IOException, TimeoutException {
​
        Connection connection = ConnUtils.getConn();
​
        Channel channel = connection.createChannel();
​
        String queueName = "dlx.queue";
        String exchangeName = "dlx.exchange";
        String routingKey = "#";
​
        // 申明exchange
        channel.exchangeDeclare(exchangeName, "topic");
        // 申明队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 队列绑定到exchange
        channel.queueBind(queueName, exchangeName, routingKey, null);
​
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {
                    System.out.println("---死信队列消费者---");
                    System.out.println(new String(message.getBody(), "UTF-8"));
                } finally {
                    // consumer手动 ack 给broker
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }
            }
        };
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("---消费者:cancelCallback");
            }
        };
​
        // 消费消息,autoAck一定要设置为false
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback);
    }
}

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

Vue3新组件transition(动画过渡)

transition组件&#xff1a;控制V-if与V-show的显示与隐藏动画 1.基本使用 <template><div><button click"falg !falg">切换</button><transition name"fade" :enter-to-class"etc"><div v-if"falg&quo…

为什么git有些commit记录,只有git reflog可以看到,git log看不到?

文章目录 原因分析1. git log 只能显示 **可达的** 提交2. git reflog 记录所有引用的变更 常见导致 git log 看不到提交的原因1. git reset 操作2. git rebase 操作3. 分支删除4. git commit --amend5. 垃圾回收&#xff08;GC&#xff09;* 如何恢复 git log 看不到的提交&am…

数据库系统基础概述

文章目录 前言一、数据库基础概念 1.数据库系统的组成2.数据模型3.数据库的体系结构二、MySQL数据库 1.了解MySQL2.MySQL的特性3.MySQL的应用场景总结 前言 MySQL数据库是一款完全免费的产品&#xff0c;用户可以直接从网上下载使用&#xff0c;不用花费任何费用。这点对于初学…

多语言长文本 AI 关键字提取 API 数据接口

多语言长文本 AI 关键字提取 API 数据接口 AI / 文本 专有模型极速提取 多语言长文本 / 实时语料库。 1. 产品功能 支持长文本关键词提取&#xff1b;多语言关键词识别&#xff1b;基于 AI 模型&#xff0c;提取精准关键词&#xff1b;全接口支持 HTTPS&#xff08;TLS v1.0 …

CentOS7更换阿里云yum更新源

目前CentOS内置的更新安装源经常报错无法更新&#xff0c;或者速度不够理想&#xff0c;这个时候更换国内的镜像源就是一个不错的选择。 备份内置更新源 mv /etc/yum.repos.d/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo.backup 下载阿里云repo源&#xff08;需要系统…

后台数据管理系统 - 项目架构设计-Vue3+axios+Element-plus(0916)

接口文档: https://apifox.com/apidoc/shared-26c67aee-0233-4d23-aab7-08448fdf95ff/api-93850835 接口根路径&#xff1a; http://big-event-vue-api-t.itheima.net 本项目的技术栈 本项目技术栈基于 ES6、vue3、pinia、vue-router 、vite 、axios 和 element-plus http:/…

LeetCode 每周算法 6(图论、回溯)

LeetCode 每周算法 6&#xff08;图论、回溯&#xff09; 图论算法&#xff1a; class Solution: def dfs(self, grid: List[List[str]], r: int, c: int) -> None: """ 深度优先搜索函数&#xff0c;用于遍历并标记与当前位置(r, c)相连的所有陆地&…

HTML讲解(二)head部分

目录 1. 2.的使用 2.1 charset 2.2 name 2.2.1 describe关键字 2.2.2 keywords关键字 2.2.3 author关键字 2.2.4 http-equiv 小心&#xff01;VS2022不可直接接触&#xff0c;否则&#xff01;没这个必要&#xff0c;方源面色淡然一把抓住&#xff01;顷刻炼化&#x…

暴力枚举算法

《啊哈&#xff01;算法》学习笔记 本博客的题目仅用暴力枚举&#xff0c;并不一定是最好的解法&#xff0c;主要是了解枚举算法 例题一&#xff1a;两方框奥数 在两个方框内填入相同的数字使得等式成立&#xff1a; 代码如下&#xff1a; for(i1;i<9;i) {if((i*103)*652…

yolov8模型在Xray图像中关键点检测识别中的应用【代码+数据集+python环境+GUI系统】

yolov8模型在X yolov8模型在Xray图像中关键点检测识别中的应用【代码数据集python环境GUI系统】 1.背景意义 X射线是一种波长极短、穿透能力极强的电磁波。当X射线穿透物体时&#xff0c;不同密度和厚度的物质会吸收不同程度的X射线&#xff0c;从而在接收端产生不同强度的信号…

Python办公自动化教程(003):PDF的加密

【1】代码 from PyPDF2 import PdfReader, PdfWriter# 读取PDF文件 pdf_reader PdfReader(./file/Python教程_1.pdf) pdf_writer PdfWriter()# 对第1页进行加密 page pdf_reader.pages[0]pdf_writer.add_page(page) # 设置密码 pdf_writer.encrypt(3535)with open(./file/P…

Google 扩展 Chrome 安全和隐私功能

过去一周&#xff0c;谷歌一直在推出新特性和功能&#xff0c;旨在让用户在 Chrome 上的桌面体验更加安全&#xff0c;最新的举措是扩展在多个设备上保存密钥的功能。 到目前为止&#xff0c;Chrome 网络用户只能将密钥保存到 Android 上的 Google 密码管理器&#xff0c;然后…

240912-设置WSL中的Ollama可在局域网访问

A. 最终效果 B. 设置Ollama&#xff08;前提&#xff09; sudo vim /etc/systemd/system/ollama.service[Unit] DescriptionOllama Service Afternetwork-online.target[Service] ExecStart/usr/bin/ollama serve Userollama Groupollama Restartalways RestartSec3 Environme…

基于SpringBoot+Vue的时尚美妆电商网站系统

作者&#xff1a;计算机学姐 开发技术&#xff1a;SpringBoot、SSM、Vue、MySQL、JSP、ElementUI、Python、小程序等&#xff0c;“文末源码”。 专栏推荐&#xff1a;前后端分离项目源码、SpringBoot项目源码、SSM项目源码 精品专栏&#xff1a;Java精选实战项目源码、Python精…

STM32 单片机最小系统全解析

STM32 单片机最小系统全解析 本文详细介绍了 STM32 单片机最小系统&#xff0c;包括其各个组成部分及设计要点与注意事项。STM32 最小系统在嵌入式开发中至关重要&#xff0c;由电源、时钟、复位、调试接口和启动电路等组成。 在电源电路方面&#xff0c;采用 3.3V 直流电源供…

(已解决)vscode如何传入argparse参数来调试/运行python程序

文章目录 前言调试传入参数运行传入参数延申 前言 以前&#xff0c;我都是用Pycharm专业版的&#xff0c;由于其好像在外网的时候&#xff0c;不能够通过VPN来连接内网服务器&#xff0c;我就改用了vscode。改用了之后&#xff0c;遇到一个问题&#xff0c;调试或者运行python…

解密.bixi、.baxia勒索病毒:如何安全恢复被加密数据

导言 在数字化时代&#xff0c;数据安全已成为个人和企业面临的重大挑战之一。随着网络攻击手段的不断演进&#xff0c;勒索病毒的出现尤为引人关注。其中&#xff0c;.bixi、.baxia勒索病毒是一种新型的恶意软件&#xff0c;它通过加密用户的重要文件&#xff0c;迫使受害者支…

redis基本数据结构-sorted set

1. sorted set的简单介绍 参考链接&#xff1a;https://mp.weixin.qq.com/s/srkd73bS2n3mjIADLVg72A Redis的Sorted Set&#xff08;有序集合&#xff09;是一种数据结构&#xff0c;它是一个不重复的字符串集合&#xff0c;每个元素都有一个对应的分数&#xff08;score&…

科研入门学习

学习视频链接 为什么要读论文 读哪些论文 论文的分类 论文质量 如何找论文 根据领域大牛的名字进行搜索查看高水平论文引用的论文&#xff0c;高水平论文引用的论文很大程度也是高水平的论文 如何整理论文 如何读论文 读论文的困境 不同人群阅读差异 读论文的方式 论文的结构…

Day.js时间插件的安装引用与常用方法大全

&#x1f680; 个人简介&#xff1a;某大型国企资深软件研发工程师&#xff0c;信息系统项目管理师、CSDN优质创作者、阿里云专家博主&#xff0c;华为云云享专家&#xff0c;分享前端后端相关技术与工作常见问题~ &#x1f49f; 作 者&#xff1a;码喽的自我修养&#x1f9…