RabiitMQ延迟队列(死信交换机)

Dead Letter Exchange(死信交换机)

        在MQ中,当消息成为死信(Dead message 死掉的信息)后,消息中间件可以将其从当前队列发送到另一个队列中,这个队列就是死信队列。而 在RabbitMQ中,由于有交换机的概念,实际是将死信发送给了死信交换机(Dead Letter Exchange,简称DLX)。死信交换机和死信队列和普通的没有区别。

消息成为死信的情况

  1. 队列消息长度到达限制
  2. 消费者拒签消息,并且不把消息重新放入原队列
  3. 消息到达存活时间未被消费

有些队列的消息成为死信后,(比如过期了或者队列满了)这些死信一般情况下是会被 RabbitMQ 清理的。但是你可以配置某个交换机为此队列的死信交换机,该队列的消息成为死信后会被重新发送到此 DLX 。至于怎么处理这个DLX中的死信就是看具体的业务场景了,DLX 中的信息可以被路由到新的队列。

  • 生产者 

    /*** 普通交换机绑定普通交换机** @return*/@Beanpublic Queue queueA() {//信息配置Map<String, Object> map = new HashMap<>();//message在该队列queue的存活时间最大为15秒map.put("x-message-ttl", 15000);//x-dead-letter-exchange参数是设置该队列的死信交换器(DLX)map.put("x-dead-letter-exchange", "exchangeB");//x-dead-letter-routing-key参数是给这个DLX指定路由键map.put("x-dead-letter-routing-key", "queueB");return new Queue("queueA", true, false, false, map);}@Beanpublic DirectExchange exchangeA() {return new DirectExchange("exchangeA");}@Beanpublic Binding bindingA() {return BindingBuilder.bind(queueA()).to(exchangeA()).with("queueA");}/*** 死信交换机绑定死信交换机** @return*/@Beanpublic Queue queueB() {return new Queue("queueB");}@Beanpublic DirectExchange exchangeB() {return new DirectExchange("exchangeB");}@Beanpublic Binding bindingB() {return BindingBuilder.bind(queueB()).to(exchangeB()).with("queueB");}
  •  模拟发送请求
    @RequestMapping("/send6")public String sendSix() throws JsonProcessingException {rabbitTemplate.convertAndSend("exchangeA", "queueA", "检查订单是否过期");return "🫶";}

这时我发送请求到队列queueA,并设置了15秒的延迟,将超时的信息调用到死信交换机中。在这里我是没开启消费者所有没有消费者去处理该请求的,信息在queueA队列等待15秒后将会转到死信交换机queueB队列进行处理:

延迟队列

  • 延迟队列,即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费。经典的应用场景是下单减库存。 

        根据以上结论,在rabbitmq中消费者只要接到信息就会自动确认进行处理。所以在上面并没有开启消费者,当请求時效后(如订单未支付,定时30分钟自动取消功能)我们不应该再让它正常处理,而把该请求放到死信交换机中安排对应的处理,所以我们需要打消费者自动处理请求改成手动。 

  • 如果手动确认则当消费者调用 ack、nack、reject 几种方法进行确认,手动确认可以在业务失败后进行一些操作,如果消息未被 ACK 则会发送到下一个消费者

  • 如果某个服务忘记 ACK 了,则 RabbitMQ 不会再发送数据给它,因为 RabbitMQ 认为该服务的处理能力有限

  • ACK 机制还可以起到限流作用,比如在接收到某条消息时休眠几秒钟

  • 消息确认模式有:

  1. AcknowledgeMode.NONE:自动确认
  2. AcknowledgeMode.AUTO:根据情况确认
  3. AcknowledgeMode.MANUAL:手动确认

确认消息(局部方法处理消息)

  • 默认情况下消息消费者是自动 ack (确认)消息的,如果要手动 ack(确认)则需要修改确认模式为 manual

消费者添加手动确认消息配置配置 :

spring:rabbitmq:listener:simple:acknowledge-mode: manua
  • 消费者接受消息: 
package com.ycxw.consumer.demos;import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;/*** @author 云村小威* @create 2024-01-25 15:36*/
@Component
public class DLXReceiver {@RabbitListener(queues = {"queueA"})@RabbitHandlerpublic void handlerA(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {System.out.println("已接受到队列queueA传递过来的消息:" + msg);channel.basicReject(tag, false);// 拒接消息,如果为true则拒绝后又从新回到队列被接受(循环),除非消息过期。//channel.basicAck(tag, true); 确认消息()一次性全接受,如果为false则接受一次}/*** 接受死信消息** @param msg*/@RabbitListener(queues = {"queueB"})@RabbitHandlerpublic void handlerB(String msg) {/*** ...接受到信息,去数据库处理*/System.out.println("已接受到队列queueB传递过来的消息:" + msg);}
}

 第一次进入普通队列别拒绝后,转到死信队列中处理...

需要注意的 basicAck 方法需要传递两个参数

  • deliveryTag(唯一标识 ID):当一个消费者向 RabbitMQ 注册后,会建立起一个 Channel ,RabbitMQ 会用 basic.deliver 方法向消费者推送消息,这个方法携带了一个 delivery tag, 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID,是一个单调递增的正整数,delivery tag 的范围仅限于 Channel

  • multiple:为了减少网络流量,手动确认可以被批处理,当该参数为 true 时,则可以一次性确认 delivery_tag 小于等于传入值的所有消息

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

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

相关文章

IS-IS weight影响路由加表

拓扑图 配置 nexthop weight影响路由加入路由表 weight默认为255&#xff0c;取值1~255&#xff0c;值越小越优先 sysname R1 # isis 1is-level level-1cost-style widenetwork-entity 49.1234.0000.0000.0001.00log-peer-change topology # interface GigabitEthernet0/0/0…

003集——通过VBA将二进制文件转为文本文件

对于二进制的文件&#xff0c;我们可以通过vba代码转为文本文件。这里以dxf为例&#xff0c; 代码如下&#xff1a; Sub ConvertBinaryToText()Dim fs As Object FileSystemObjectSet fs CreateObject("Scripting.FileSystemObject")Dim binaryFilePath As Strin…

Java算法练习5

Java算法练习5 1.8 [268. 丢失的数字](https://leetcode.cn/problems/missing-number/)1.9 [383. 赎金信](https://leetcode.cn/problems/ransom-note/)1.10 [2133. 检查是否每一行每一列都包含全部整数](https://leetcode.cn/problems/check-if-every-row-and-column-contains…

STM32Cubmax stm32f103zet6 SPI通讯

一、基本概念 SPI 是英语 Serial Peripheral interface 的缩写&#xff0c;顾名思义就是串行外围设备接口。是 Motorola 首先在其 MC68HCXX 系列处理器上定义的。 SPI 接口主要应用在 EEPROM&#xff0c; FLASH&#xff0c;实时时 钟&#xff0c; AD 转换器&#xff0c;还有数…

【QT】VS-code报错:LNK2019: 无法解析的外部符号

目录 0.环境 1.问题简述 2.分析报错原因 3.解决方法 1&#xff09;set() 相关语句 2&#xff09;target_link_libraries() 相关语句 4.参考 0.环境 windows11 、 vs-code 、 qt 、 c、编译器为vs2019-x86_amd64 1.问题简述 项目编译release版本时会报错&#xff1a;报错…

Linux操作系统运维-Docker的基础知识梳理总结

Linux操作系统运维-Docker的基础知识梳理总结 docker用来解决不同开发人员软件调试时环境不统一的问题&#xff0c;保证了程序调试时运行环境的一致性。docker的设计理念便是一处镜像&#xff0c;处处运行&#xff0c;即通过产生用户软件&#xff0c;运行环境及其运行配置的统一…

C++ || 模板初阶 | 函数模板 | 类模板

泛型编程 泛型编程&#xff0c;编写与类型无关的通用代码&#xff0c;是代码复用的一种手段。模板是泛型编程的基础。 可以理解为活字印刷术类似的方式。 函数模板 函数模板概念 函数模板&#xff0c;代表一个函数家族&#xff0c;该函数模板与类型无关&#xff0c;在使用…

解决dockor安装nginx提示missing signature key的问题

问题描述 使用dockor安装nginx拉取nginx的时候提示key丢失问题 问题定位 由于dockor版本低导致 问题解决 卸载重新安装最新版本dockor 解决步骤 1. 卸载旧版本的Docker&#xff1a; sudo yum remove docker docker-common docker-selinux docker-engine 2. 安装依赖包&am…

Flink Format系列(2)-CSV

Flink的csv格式支持读和写csv格式的数据&#xff0c;只需要指定 format csv&#xff0c;下面以kafka为例。 CREATE TABLE user_behavior (user_id BIGINT,item_id BIGINT,category_id BIGINT,behavior STRING,ts TIMESTAMP(3) ) WITH (connector kafka,topic user_behavior…

【人工智能】神奇的Embedding:文本变向量,大语言模型智慧密码解析(10)

什么是嵌入&#xff1f; OpenAI 的文本嵌入衡量文本字符串的相关性。嵌入通常用于&#xff1a; Search 搜索&#xff08;结果按与查询字符串的相关性排序&#xff09;Clustering 聚类&#xff08;文本字符串按相似性分组&#xff09;Recommendations 推荐&#xff08;推荐具有…

牛客网SQL:查询每个日期新用户的次日留存率

官网链接&#xff1a; 牛客每个人最近的登录日期(五)_牛客题霸_牛客网牛客每天有很多人登录&#xff0c;请你统计一下牛客每个日期新用户的次日留存率。 有一个登录(login。题目来自【牛客题霸】https://www.nowcoder.com/practice/ea0c56cd700344b590182aad03cc61b8?tpId82 …

datax离线同步oracle表到clickhouse实践1

时间&#xff1a;2024.01 目录1、安装启动 oracle19c 容器 2、rpm包安装clickhouse 3、datax安装 4、datax同步 目标库根据要同步的表&#xff0c;按照clickhouse建表规范建表 编写json文件 编写增量同步shell脚本&#xff0c;加入 crond 定时任务 1、安装启动 oracle19c 容器…