RabbitMQ (4)

RabbitMQ (4)

文章目录

  • 1. 死信的概念
  • 2. 死信的来源
  • 3. 死信代码案例
    • 3.1 TTL 过期时间
    • 3.2 超过队列最大长度
    • 3.3 拒绝消息

前言
 

上文我们已经学习完 交换机 ，知道了几个交换机的使用 ，下面我们来学习一下 死信队列

 

1. 死信的概念

 
先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息 进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
 
应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说：用户在商城下单成功并点击去支付后在指定时间未支付时自动失效。

 

2. 死信的来源

 

  1. 消息 TTL 过期 : TTL 是 Time To Live 的缩写， TTL 就是 生存时间
  2. 队列达到最长长度 : 队列满了 ， 无法添加数据到 MQ 中
  3. 消息被拒绝 （basic.reject 或 basic.nack） 并且 requeue = false

 

3. 死信代码案例

 

这里 创建一个 direct 交换机 ，两个消费者 ， 一个生产者 ， 两个 队列 (一个为 消息队列 ， 一个为死信队列)

 
图:

在这里插入图片描述

 
代码 :

 

3.1 TTL 过期时间

 

生产者：

package org.example.seven;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

// 生产者
public class Producer {

    // 普通交换机的名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        // 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 10s
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
        }
    }
}

 

消费者 c1 （启动之后关闭该消费者， 模拟其接受不到消息）

package org.example.seven;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;

public class Consumer01 {

    // 普通交换机的名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // 死信交换机的名称
    public static final String DEAD_EXCHANGE = "dead_change";

    // 普通队列的名称
    public static final String NORMAL_QUEUE = "normal_queue";

    // 死刑队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明死信和普通交换机 , 类型为 direct （直接交换机）

        // 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();

        // 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);

        // 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);

        // 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key", "lisi");

        // 声明队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

        // 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");

        // 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("等待接受消息");

        DeliverCallback deliverCallback = (tag, message) -> {
            System.out.println("C1 接收到的消息为: " + new String(message.getBody(), "UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {
        });
    }
}

 

先启动消费者 C1，创建出队列，然后停止该 C1 的运行，则 C1 将无法收到队列的消息，无法收到的消息 10 秒后进入死信队列。启动生产者 producer 生产消息

在这里插入图片描述

 
c1 看完，我们在来写 c2 消费者 ，将进入到死信队列的消息 进行消费.

 
消费者c2

package org.example.seven;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer02 {

    // 死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        System.out.println("等待接受死信消息......");

        DeliverCallback deliverCallback = (tag, message) -> {
            System.out.println("C2 接收到的消息： " + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback,(tag)->{});
    }
}

 
图:

在这里插入图片描述

 
看完 消息过期后 ，消息转发到 死信队列 被 c2 消费，下面我们来 尝试使用 死信最大长度 (队列满了，将多的消息转发到死信队列中）

 

3.2 超过队列最大长度

 

消息生产者代码 去掉 TTL 属性 ， 将 basicPublish 的第三个参数改为 null

 
生产者:

package org.example.seven;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

// 生产者
public class Producer {

    // 普通交换机的名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        // 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 10s
//        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
//            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());
        }
    }
}

 
c1 消费者 (启动之后关闭该消费者 模拟其接收不到消息)

package org.example.seven;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;

public class Consumer01 {

    // 普通交换机的名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // 死信交换机的名称
    public static final String DEAD_EXCHANGE = "dead_change";

    // 普通队列的名称
    public static final String NORMAL_QUEUE = "normal_queue";

    // 死刑队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明死信和普通交换机 , 类型为 direct （直接交换机）

        // 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();

        // 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);

        // 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);

        // 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key", "lisi");

        // 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信
        arguments.put("x-max-length", 6);

        // 声明队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

        // 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");

        // 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("等待接受消息");

        DeliverCallback deliverCallback = (tag, message) -> {
            System.out.println("C1 接收到的消息为: " + new String(message.getBody(), "UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {
        });
    }
}

 
注意:
 
这参数改变了(没有设置 ttl 时间，新增了 队列的 最大长度限制 为 6) ，所以 需要把原来队列删除

 
消费者c2 代码不变

package org.example.seven;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer02 {

    // 死信队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        System.out.println("等待接受死信消息......");

        DeliverCallback deliverCallback = (tag, message) -> {
            System.out.println("C2 接收到的消息： " + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback,(tag)->{});
    }
}

 
效果:

在这里插入图片描述

 
这里 之所以要启动 c1 后在关闭，是为了展示 6个消息放到 普通队列 ，4个消息放到死信队列， 如果不这么做，发送的 10个消息 都会被 c1 消费 ，（消息发送到 队列后 , 立马 转发给 c1 导致 队列就不会达到 6 个 ，队列不会满 ，也就不会将消息转化给 死信队列).

在这里插入图片描述

 

3.3 拒绝消息

 
消息生产者 和 消费者 c2 与上面的代码一样

这里我们 拒绝 info7 消息 ，想要 拒绝 info7 消息，我们可以采用手动应答.

 
消费者c1

package org.example.seven;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;

public class Consumer01 {

    // 普通交换机的名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // 死信交换机的名称
    public static final String DEAD_EXCHANGE = "dead_change";

    // 普通队列的名称
    public static final String NORMAL_QUEUE = "normal_queue";

    // 死刑队列的名称
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        // 声明死信和普通交换机 , 类型为 direct （直接交换机）

        // 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();

        // 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);

        // 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);

        // 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key", "lisi");

        // 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信
//        arguments.put("x-max-length", 6);

        // 声明队列
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);

        // 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");

        // 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("等待接受消息");

        DeliverCallback deliverCallback = (tag, message) -> {
            String msg = new String(message.getBody(), "UTF-8");
            if (msg.equals("info7")) {
                System.out.println("C1 接收到消息为: " + msg + " 此消息被 C1 拒绝");
                //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            } else {
                System.out.println("C1 接收到的消息为: " + msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
//        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {});
        // 开启 手动应答
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, (tag) -> {
        });
    }
}

在这里插入图片描述

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

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

相关文章

BUUCTF刷题记录

BUUCTF刷题记录

[BJDCTF2020]Easy MD51 进入题目页面&#xff0c;题目提示有一个链接&#xff0c;应该是题目源码 进入环境&#xff0c;是一个查询框&#xff0c;无论输入什么都没有回显&#xff0c;查看源码也没什么用 利用bp抓包查看有没有什么有用的东西 发现响应的Hint那里有一个sql语句&…
阅读更多...
WIN11新版画图问题解决

WIN11新版画图问题解决

1 白色背景被连同删除的问题 解决方法&#xff1a;加层 将层调整为新建的层&#xff0c;在这个层下画图就行。 2 QQ截图无法直接放在画图上的问题 使用QQ截图的时候&#xff1a; 解决方法&#xff1a;使用windows自带的截图工具 步骤&#xff1a; 1. 使用快捷键winshifts 2.…
阅读更多...
【Git企业开发】第一节.Git 初识

【Git企业开发】第一节.Git 初识

作者简介&#xff1a;大家好&#xff0c;我是未央&#xff1b; 博客首页&#xff1a;未央.303 系列专栏&#xff1a; 每日一句&#xff1a;人的一生&#xff0c;可以有所作为的时机只有一次&#xff0c;那就是现在&#xff01;&#xff01;&#xff01;&#xff01;&#xff01…
阅读更多...
Remote Local File Inclusion (RFI/LFI)-文件包含漏洞

Remote Local File Inclusion (RFI/LFI)-文件包含漏洞

在Web应用开发过程中，程序开发者经常会把具有某一功能的部分代码封装起来形成独立的文件，在后续想实现该功能时，就不需要重复编写，直接调用文件，大大提高编程效率。这种调用文件的过程一般被称为文件包含。开发人员为了使代码更灵活，会将被包含的文件设置为变量，用来进行…
阅读更多...
1.1 计算机安全概念

1.1 计算机安全概念

思维导图&#xff1a; 前言&#xff1a; 第1章: 计算机与网络安全概念笔记 1. 学习目标 了解保密性、完整性和可用性的关键安全需求。了解OSI的X.800安全架构。识别和举例说明不同的安全威胁和攻击。掌握安全设计的基本准则。熟悉攻击面和攻击树的使用。了解与密码标准相关的…
阅读更多...
EASYX剪切区域

EASYX剪切区域

eg1:EASY中的颜色模型 可以参考推荐16进制颜色表&#xff1a;https://www.codeeeee.com/color/rgb.html 参考学习EASYX在线文档https://docs.easyx.cn/zh-cn/drawing-func easyx的基本概念和使用方式 #include <stdio.h> #include <easyx.h> #include <iostr…
阅读更多...
Winform 多语言化快速解析替换工具-1分钟一个界面

Winform 多语言化快速解析替换工具-1分钟一个界面

随着业务的扩展&#xff0c;有的软件有多语言化的需求。那么如果软件已经很多写死的文字内容如何快速进行语言化替换呢&#xff0c;一个一个去改工作量太大。 于是开发了个小工具用来替换现有内容并生成语音包&#xff0c;原理就是采用正则表达式进行匹配控件关键字以及中文进…
阅读更多...
orb-slam3编译手册（Ubuntu20.04）

orb-slam3编译手册（Ubuntu20.04）

orb-slam3编译手册&#xff08;Ubuntu20.04&#xff09; 一、环境要求1.安装git2.安装g3.安装CMake4.安装vi编辑器 二、源代码下载三、依赖库下载1.Eigen安装2.Pangolin安装3.opencv安装4.安装Python & libssl-dev5.安装boost库 三、安装orb-slam3四、数据集下载及测试 写在…
阅读更多...
关于线性模型的底层逻辑解读 (机器学习 细读01)

关于线性模型的底层逻辑解读 (机器学习 细读01)

一 多元线性回归 线性回归是机器学习中 有监督机器学习 下的一种算法。 回归问题主要关注的是因变量(需要预测的值&#xff0c;可以是一个也可以是多个)和一个或多个数值型的自变量(预测变量)之间的关系。 需要预测的值:即目标变量&#xff0c;target&#xff0c;y&#xff0c…
阅读更多...
计算机网络重点概念整理-第六章 应用层【期末复习|考研复习】

计算机网络重点概念整理-第六章 应用层【期末复习|考研复习】

第六章 应用层 【期末复习|考研复习】 计算机网络系列文章传送门&#xff1a; 第一章 计算机网络概述 第二章 物理层 第三章 数据链路层 第四章 网络层 第五章 传输层 第六章 应用层 第七章 网络安全 计算机网络整理-简称&缩写 文章目录 第六章 应用层 【期末复习|考研复习…
阅读更多...
爬取抖音用户的个人基本信息

爬取抖音用户的个人基本信息

今年夏季&#xff0c;大概七八月份&#xff0c;刀郎开通抖音账号&#xff0c;并在抖音上发布多首作品&#xff0c;一时之间其热度暴涨&#xff0c;其粉丝也是与日俱增。 有人为了蹭热度&#xff0c;直播刀郎粉丝的实时变化情况&#xff0c;直播热度最高的时候同时几千人在线观…
阅读更多...
HttpClient远程使用大全

HttpClient远程使用大全

一 HttpClient简介 1.1 概述 HttpClient只能以编程的方式通过其API用于传输和接受HTTP消息。主要实现功能&#xff1a; 实现了所有 HTTP 的方法&#xff08;GET、POST、PUT、HEAD、DELETE、HEAD、OPTIONS 等&#xff09; 支持 HTTPS 协议 支持代理服务器&#xff08;Nginx…
阅读更多...
AS/400简介

AS/400简介

AS400 AS400 简介AS/400操作系统演示 AS400 简介 在 AS400 中&#xff0c;AS代表“应用系统”。它是多用户、多任务和非常安全的系统&#xff0c;因此用于需要同时存储和处理敏感数据的行业。它最适合中级行业&#xff0c;因此用于制药行业、银行、商场、医院管理、制造业、分销…
阅读更多...
Web APIs——事件流

Web APIs——事件流

一、事件流 1.1 事件流与两个阶段说明 事件流指的是事件完整执行过程中的流动路径 说明&#xff1a;假设页面里有个div&#xff0c;当触发事件时&#xff0c;会经历两个阶段&#xff0c;分别是捕获阶段、冒泡阶段 简单来说&#xff1a;捕获阶段是 从父到子 冒泡阶段是从子到父…
阅读更多...
Linux网络编程01

Linux网络编程01

网络层级 协议 协议&#xff1a;两个对等实体对通话内容的约定&#xff0c;一个协议是对应收发双方相同层级的 常见的协议 应用层&#xff08;公开协议&#xff09;&#xff1a; http协议&#xff08;浏览网页&#xff09;&#xff1b;客户端&#xff08;浏览器&#xff09;发…
阅读更多...
【206.反转链表】

【206.反转链表】

目录 一、题目描述二、算法原理三、代码实现 一、题目描述 二、算法原理 三、代码实现 class Solution { public:ListNode* reverseList(ListNode* head) {if(headnullptr) return nullptr;if(head->nextnullptr) return head;ListNode* newheadreverseList(head->next)…
阅读更多...
一款功能强大的iOS设备管理软件Mazing 3中文版免费2024最新下载

一款功能强大的iOS设备管理软件Mazing 3中文版免费2024最新下载

Mazing 3中文版是一款功能强大的iOS设备管理软件&#xff0c;它可以帮助用户备份和管理他们的iPhone、iPad或iPod Touch上的数据。除此之外&#xff0c;它还可以将备份数据转移到新的设备中、管理应用程序、导入和导出媒体文件等。本文将详细介绍iMazing的功能和安全性&#xf…
阅读更多...
Unity URP14.0 自定义后处理框架

Unity URP14.0 自定义后处理框架

目录 碎碎念一些基础CustomPostProcessing.csCustomPostProcessingFeature.csCustomPostProcessingPass.cs例子&#xff1a;BSC后处理shader&#xff08;BSC&#xff09;后处理cs脚本(BSC) 例子&#xff1a;ColorBlitPostProcessing.hlslColorBlit2.shaderColorBlit.cs文件 其他…
阅读更多...
震惊！ 全方位解释在测试眼里，什么是需求？为什么要有需求？深入理解需求——图文并茂，生活举例，简单好理解

震惊！ 全方位解释在测试眼里，什么是需求？为什么要有需求？深入理解需求——图文并茂，生活举例，简单好理解

1、什么是需求&#xff1f; 需求定义(官方) 满足用户期望或正式规定文档&#xff08;合同、标准、规范&#xff09;所具有的条件和权能&#xff0c;包含用户需求和软件需求 用户需求&#xff1a;可以简单理解为甲方提出的需求&#xff0c;如果没有甲方&#xff0c;那么就是终端…
阅读更多...
计算机网络【CN】子网划分与子网掩码

计算机网络【CN】子网划分与子网掩码

一个子网定义(X.X.X.X/n) 子网掩码为 n 个 1&#xff0c;32-n 个 0包含的 IP 地址数&#xff1a;232−n 主机号全 0 表示本网段主机号全 1 表示网段的广播地址可分配的 IP 地址数 :232−&#x1d45b;−2 子网划分原则 满足子网定义子网&#x1d434;1…&#x1d434;&#x…
阅读更多...
最新文章

Copyright @ 2022~2023