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

2024-02-07 09:12

本文主要是介绍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 小于等于传入值的所有消息

这篇关于RabiitMQ延迟队列(死信交换机)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

hdu1180(广搜+优先队列)

此题要求最少到达目标点T的最短时间,所以我选择了广度优先搜索,并且要用到优先队列。 另外此题注意点较多,比如说可以在某个点停留,我wa了好多两次,就是因为忽略了这一点,然后参考了大神的思想,然后经过反复修改才AC的 这是我的代码 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<

poj 3190 优先队列+贪心

题意: 有n头牛,分别给他们挤奶的时间。 然后每头牛挤奶的时候都要在一个stall里面,并且每个stall每次只能占用一头牛。 问最少需要多少个stall,并输出每头牛所在的stall。 e.g 样例: INPUT: 51 102 43 65 84 7 OUTPUT: 412324 HINT: Explanation of the s

poj 2431 poj 3253 优先队列的运用

poj 2431: 题意: 一条路起点为0, 终点为l。 卡车初始时在0点,并且有p升油,假设油箱无限大。 给n个加油站,每个加油站距离终点 l 距离为 x[i],可以加的油量为fuel[i]。 问最少加几次油可以到达终点,若不能到达,输出-1。 解析: 《挑战程序设计竞赛》: “在卡车开往终点的途中,只有在加油站才可以加油。但是,如果认为“在到达加油站i时,就获得了一

poj3750约瑟夫环,循环队列

Description 有N个小孩围成一圈,给他们从1开始依次编号,现指定从第W个开始报数,报到第S个时,该小孩出列,然后从下一个小孩开始报数,仍是报到S个出列,如此重复下去,直到所有的小孩都出列(总人数不足S个时将循环报数),求小孩出列的顺序。 Input 第一行输入小孩的人数N(N<=64) 接下来每行输入一个小孩的名字(人名不超过15个字符) 最后一行输入W,S (W < N),用

POJ2010 贪心优先队列

c头牛,需要选n头(奇数);学校总共有f的资金, 每头牛分数score和学费cost,问合法招生方案中,中间分数(即排名第(n+1)/2)最高的是多少。 n头牛按照先score后cost从小到大排序; 枚举中间score的牛,  预处理左边与右边的最小花费和。 预处理直接优先队列贪心 public class Main {public static voi

Java并发编程之——BlockingQueue(队列)

一、什么是BlockingQueue BlockingQueue即阻塞队列,从阻塞这个词可以看出,在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种: 1. 当队列满了的时候进行入队列操作2. 当队列空了的时候进行出队列操作123 因此,当一个线程试图对一个已经满了的队列进行入队列操作时,它将会被阻塞,除非有另一个线程做了出队列操作;同样,当一个线程试图对一个空

ActiveMQ—消息特性(延迟和定时消息投递)

ActiveMQ消息特性:延迟和定时消息投递(Delay and Schedule Message Delivery) 转自:http://blog.csdn.net/kimmking/article/details/8443872 有时候我们不希望消息马上被broker投递出去,而是想要消息60秒以后发给消费者,或者我们想让消息没隔一定时间投递一次,一共投递指定的次数。。。 类似

FreeRTOS学习笔记(六)队列

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档 文章目录 前言一、队列的基本内容1.1 队列的引入1.2 FreeRTOS 队列的功能与作用1.3 队列的结构体1.4 队列的使用流程 二、相关API详解2.1 xQueueCreate2.2 xQueueSend2.3 xQueueReceive2.4 xQueueSendFromISR2.5 xQueueRecei

计算机网络基础概念 交换机、路由器、网关、TBOX

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档 文章目录 前言一、VLAN是什么?二 、交换机三、路由器四、网关五、TBOXTelematics BOX,简称车载T-BOX,车联网系统包含四部分,主机、车载T-BOX、手机APP及后台系统。主机主要用于车内的影音娱乐,以及车辆信息显示;车载T-BOX主要用于和后台系统/手机APP通信,实现手机APP的车辆信息显示与控

MySQL主从同步延迟原理及解决方案

概述 MySQL的主从同步是一个很成熟的架构,优点为: ①在从服务器可以执行查询工作(即我们常说的读功能),降低主服务器压力; ②在从主服务器进行备份,避免备份期间影响主服务器服务; ③当主服务器出现问题时,可以切换到从服务器。 相信大家对于这些好处已经非常了解了,在项目的部署中也采用这种方案。但是MySQL的主从同步一直有从库延迟的问题,那么为什么会有这种问题。这种问题如何解决呢? MyS