如何保证消息不丢失?——使用rabbitmq的死信队列!

2024-04-10 23:52

本文主要是介绍如何保证消息不丢失?——使用rabbitmq的死信队列!,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

如何保证消息不丢失?——使用rabbitmq的死信队列!

1、什么是死信

在 RabbitMQ 中充当主角的就是消息,在不同场景下,消息会有不同地表现。
死信就是消息在特定场景下的一种表现形式,这些场景包括:

    1. 消息被拒绝访问,即 RabbitMQ返回 basicNack 的信号时 或者拒绝basicReject
    1. 消费者发生异常,超过重试次数 。 其实spring框架调用的就是 basicNack
    1. 消息的Expiration 过期时长或队列TTL过期时间。
    1. 消息队列达到最大容量

上述场景经常产生死信,即消息在这些场景中时,被称为死信。

2、什么是死信队列

死信队列就是用于储存死信的消息队列,在死信队列中,有且只有死信构成,不会存在其余类型的消息。
死信队列在 RabbitMQ 中并不会单独存在,往往死信队列都会绑定这一个普通的业务消息队列,当所绑定的消息队列中,有消息变成死信了,那么这个消息就会重新被交换机路由到指定的死信队列中去,我们可以通过对这个死信队列进行监听,从而手动的去对这一消息进行补偿。 人工干预
在这里插入图片描述

3、那么,我们到底如何来使用死信队列呢?

死信队列基本使用,只需要在声明业务队列的时候,绑定指定的死信交换机和RoutingKey即可。

生产者

/** Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.**/
package com.fpl.provider;import com.fpl.model.OrderingOk;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;/*** <p>Project: spring-rabbitmq - DeadProvider</p>* <p>Powered by fpl1116 On 2024-04-09 11:35:12</p>* <p>描述:<p>** @author penglei* @version 1.0* @since 1.8*/
@Service
public class DeadProvider {@Autowiredprivate RabbitTemplate rabbitTemplate;public void send(OrderingOk orderingOk) {rabbitTemplate.convertAndSend("Direct_E01", "RK01", orderingOk,new MessagePostProcessor(){@Overridepublic Message postProcessMessage(Message message) throws AmqpException {int id  = orderingOk.getId();int expiration = 0;if(id == 1){expiration = 50*1000;}else if(id == 2){expiration = 40*1000;}else if(id ==3){expiration = 30*1000;}else if(id ==4){expiration = 20*1000;}else if(id ==5){expiration = 10*1000;}//为每个消息设置过期时长,但是有可能造成最前面的一个消息未过期一直阻塞后面的消息不能被消费message.getMessageProperties().setExpiration(String.valueOf(expiration));return message;}});}
}

消费者

/** Copyright (c) 2020, 2024, fpl1116.cn All rights reserved.**/
package com.fpl.consumers;import com.fpl.model.OrderingOk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;/*** <p>Project: spring-rabbitmq - DeadConsumer</p>* <p>Powered by fpl1116 On 2024-04-09 11:32:59</p>* <p>描述:<p>** @author penglei* @version 1.0* @since 1.8*/
//@Configuration
@Slf4j
public class DeadConsumer {//死信交换机@Beanpublic DirectExchange deadExchange(){return  ExchangeBuilder.directExchange("Dead_E01").build();}//死信队列@Beanpublic Queue deadQueue1(){return   QueueBuilder.durable("Dead_Q01").build();}//死信交换机与死信队列的绑定@Beanpublic Binding deadBinding1(Queue deadQueue1,DirectExchange deadExchange){return BindingBuilder.bind(deadQueue1).to(deadExchange).with("RK_DEAD");}//业务队列@Beanpublic Queue queue1(){return   QueueBuilder.durable("Direct_Q01").deadLetterExchange("Dead_E01").deadLetterRoutingKey("RK_DEAD")//.ttl(10*1000) //该属性是队列的属性,设置消息的过期时间,消息在队列里面停留时间n毫秒后,就会把这个消息投递到死信交换机,针对的是所有的消息//.maxLength(20) //设置队列存放消息的最大个数,x-max-length属性值,当队列里面消息超过20,会把队列之前的消息依次放进死信队列.build();}//业务交换机@Beanpublic DirectExchange exchange(){return  ExchangeBuilder.directExchange("Direct_E01").build();}//业务交换机与队列的绑定@Beanpublic Binding binding1(Queue queue1,DirectExchange exchange){return BindingBuilder.bind(queue1).to(exchange).with("RK01");}//    @RabbitListener(queues = "Direct_Q01")
//    public void receiveMessage(OrderingOk msg,Message message, Channel channel) throws IOException {
//
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
//
//        System.out.println("消费者1 收到消息:"+ msg +" tag:"+deliveryTag);
//
//        channel.basicReject(deliveryTag, false);
//        try {
//            // 处理消息...
//            int  i= 5/0;
//            // 如果处理成功,手动发送ack确认 ,Yes
//            channel.basicAck(deliveryTag, false);
//        } catch (Exception e) {
//            // 处理失败,可以选择重试或拒绝消息(basicNack或basicReject)  NO
//            channel.basicNack(deliveryTag, false, false); // 并重新入队
//
//        }
}//}

测试

@Testvoid test4() throws IOException {for (int i = 1; i <=5;i++){OrderingOk orderingOk = OrderingOk.builder().id(i).name("fpl " + i).build();deadProvider.send(orderingOk);System.out.println("发送成功:"+i);}System.in.read();}

在这里插入图片描述

这篇关于如何保证消息不丢失?——使用rabbitmq的死信队列!的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/892506

相关文章

如何使用 Python 读取 Excel 数据

《如何使用Python读取Excel数据》:本文主要介绍使用Python读取Excel数据的详细教程,通过pandas和openpyxl,你可以轻松读取Excel文件,并进行各种数据处理操... 目录使用 python 读取 Excel 数据的详细教程1. 安装必要的依赖2. 读取 Excel 文件3. 读

解决Maven项目idea找不到本地仓库jar包问题以及使用mvn install:install-file

《解决Maven项目idea找不到本地仓库jar包问题以及使用mvninstall:install-file》:本文主要介绍解决Maven项目idea找不到本地仓库jar包问题以及使用mvnin... 目录Maven项目idea找不到本地仓库jar包以及使用mvn install:install-file基

Python使用getopt处理命令行参数示例解析(最佳实践)

《Python使用getopt处理命令行参数示例解析(最佳实践)》getopt模块是Python标准库中一个简单但强大的命令行参数处理工具,它特别适合那些需要快速实现基本命令行参数解析的场景,或者需要... 目录为什么需要处理命令行参数?getopt模块基础实际应用示例与其他参数处理方式的比较常见问http

JAVA保证HashMap线程安全的几种方式

《JAVA保证HashMap线程安全的几种方式》HashMap是线程不安全的,这意味着如果多个线程并发地访问和修改同一个HashMap实例,可能会导致数据不一致和其他线程安全问题,本文主要介绍了JAV... 目录1. 使用 Collections.synchronizedMap2. 使用 Concurren

C 语言中enum枚举的定义和使用小结

《C语言中enum枚举的定义和使用小结》在C语言里,enum(枚举)是一种用户自定义的数据类型,它能够让你创建一组具名的整数常量,下面我会从定义、使用、特性等方面详细介绍enum,感兴趣的朋友一起看... 目录1、引言2、基本定义3、定义枚举变量4、自定义枚举常量的值5、枚举与switch语句结合使用6、枚

Java的栈与队列实现代码解析

《Java的栈与队列实现代码解析》栈是常见的线性数据结构,栈的特点是以先进后出的形式,后进先出,先进后出,分为栈底和栈顶,栈应用于内存的分配,表达式求值,存储临时的数据和方法的调用等,本文给大家介绍J... 目录栈的概念(Stack)栈的实现代码队列(Queue)模拟实现队列(双链表实现)循环队列(循环数组

使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)

《使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)》PPT是一种高效的信息展示工具,广泛应用于教育、商务和设计等多个领域,PPT文档中常常包含丰富的图片内容,这些图片不仅提升了... 目录一、引言二、环境与工具三、python 提取PPT背景图片3.1 提取幻灯片背景图片3.2 提取

使用Python实现图像LBP特征提取的操作方法

《使用Python实现图像LBP特征提取的操作方法》LBP特征叫做局部二值模式,常用于纹理特征提取,并在纹理分类中具有较强的区分能力,本文给大家介绍了如何使用Python实现图像LBP特征提取的操作方... 目录一、LBP特征介绍二、LBP特征描述三、一些改进版本的LBP1.圆形LBP算子2.旋转不变的LB

Maven的使用和配置国内源的保姆级教程

《Maven的使用和配置国内源的保姆级教程》Maven是⼀个项目管理工具,基于POM(ProjectObjectModel,项目对象模型)的概念,Maven可以通过一小段描述信息来管理项目的构建,报告... 目录1. 什么是Maven?2.创建⼀个Maven项目3.Maven 核心功能4.使用Maven H

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结