RabbitMQ 实验消费原始队列消息, 拒绝(reject)投递死信交换机过程

本文主要是介绍RabbitMQ 实验消费原始队列消息, 拒绝(reject)投递死信交换机过程,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

如果你想通过 RabbitMQ 的死信队列功能实现消费者拒绝消息投递到死信交换机的行为,你可以按照以下步骤操作:

  1. 创建原始队列,并将其绑定到一个交换机上:
export RABBITMQ_SERVER=127.0.0.1
export RABBITMQ_PORT=5672   
export RABBITMQ_USER=mingcai     
export RABBITMQ_PASSWORD=passwordrabbitmqadmin list exchanges
rabbitmqadmin declare queue name=my_queue durable=true
rabbitmqadmin declare exchange name=my_exchange type=direct
rabbitmqadmin declare binding source=my_exchange destination=my_queue routing_key=my_routing_key
  1. 创建死信交换机和死信队列,并将死信队列绑定到死信交换机:
rabbitmqadmin declare exchange name=my_dlx type=direct
rabbitmqadmin declare queue name=my_dlq durable=true
rabbitmqadmin declare binding source=my_dlx destination=my_dlq routing_key=dlx_routing_key
  1. 设置原始队列的参数,包括死信交换机和死信路由键:
rabbitmqadmin declare queue name=my_queue durable=true arguments='{"x-dead-letter-exchange":"my_dlx", "x-dead-letter-routing-key":"dlx_routing_key"}'
  1. 编写消费者代码,在消费者代码中拒绝投递消息到死信交换机:
//Consumer.ts
import * as amqp from 'amqplib';async function main() {const connection = await amqp.connect('amqp://mingcai:password@127.0.0.1');const channel = await connection.createChannel();const queue = 'my_queue';await channel.consume(queue, async (msg) => {if (msg !== null) {try {await new Promise(resolve => {setTimeout(()=>{resolve(true)},3000)})// Process the messageconsole.log("Received message:", msg.content.toString());// Simulate rejection of message delivery to dead letter exchangethrow new Error("Failed to process message, rejecting delivery to dead letter exchange");} catch (error) {// Reject the message delivery to dead letter exchangechannel.reject(msg, false);console.error("Error processing message:", error.message);}}});console.log('Waiting for messages...');
}main().catch(console.error);

在这里插入图片描述

在这个示例中,当消费者处理消息时发生错误时,它会将消息的投递拒绝到死信交换机。这样,这些被拒绝的消息将被重新路由到死信队列 my_dlq 中。

记得根据你的具体需求修改队列、交换机和消费者的配置,确保它们符合你的预期行为。

rabbitmqadmin purge queue name=my_dlq #删除队列中的所有消息

//Producer.ts 
import * as amqp from 'amqplib';async function sendMsg() {try {const connection = await amqp.connect('amqp://mingcai:password@127.0.0.1');const channel = await connection.createChannel(); // 创建通道const delayQueue = 'my_queue'; // 延迟队列名称// 每秒发送一条消息for (let i = 0; i < 100; i++) {await new Promise(resolve => {setTimeout(() => {resolve(true)}, 1000)})const message = 'Hello RabbitMQ!';channel.sendToQueue(delayQueue, Buffer.from(message));console.log('消息已发送:', message);}} catch (error) {console.error('错误:', error);}
}sendMsg()

这篇关于RabbitMQ 实验消费原始队列消息, 拒绝(reject)投递死信交换机过程的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

将Mybatis升级为Mybatis-Plus的详细过程

《将Mybatis升级为Mybatis-Plus的详细过程》本文详细介绍了在若依管理系统(v3.8.8)中将MyBatis升级为MyBatis-Plus的过程,旨在提升开发效率,通过本文,开发者可实现... 目录说明流程增加依赖修改配置文件注释掉MyBATisConfig里面的Bean代码生成使用IDEA生

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整

C# WinForms存储过程操作数据库的实例讲解

《C#WinForms存储过程操作数据库的实例讲解》:本文主要介绍C#WinForms存储过程操作数据库的实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、存储过程基础二、C# 调用流程1. 数据库连接配置2. 执行存储过程(增删改)3. 查询数据三、事务处

JSON Web Token在登陆中的使用过程

《JSONWebToken在登陆中的使用过程》:本文主要介绍JSONWebToken在登陆中的使用过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录JWT 介绍微服务架构中的 JWT 使用结合微服务网关的 JWT 验证1. 用户登录,生成 JWT2. 自定义过滤

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优

SpringIntegration消息路由之Router的条件路由与过滤功能

《SpringIntegration消息路由之Router的条件路由与过滤功能》本文详细介绍了Router的基础概念、条件路由实现、基于消息头的路由、动态路由与路由表、消息过滤与选择性路由以及错误处理... 目录引言一、Router基础概念二、条件路由实现三、基于消息头的路由四、动态路由与路由表五、消息过滤

java中使用POI生成Excel并导出过程

《java中使用POI生成Excel并导出过程》:本文主要介绍java中使用POI生成Excel并导出过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录需求说明及实现方式需求完成通用代码版本1版本2结果展示type参数为atype参数为b总结注:本文章中代码均为

SpringCloud之LoadBalancer负载均衡服务调用过程

《SpringCloud之LoadBalancer负载均衡服务调用过程》:本文主要介绍SpringCloud之LoadBalancer负载均衡服务调用过程,具有很好的参考价值,希望对大家有所帮助,... 目录前言一、LoadBalancer是什么?二、使用步骤1、启动consul2、客户端加入依赖3、以服务

Oracle存储过程里操作BLOB的字节数据的办法

《Oracle存储过程里操作BLOB的字节数据的办法》该篇文章介绍了如何在Oracle存储过程中操作BLOB的字节数据,作者研究了如何获取BLOB的字节长度、如何使用DBMS_LOB包进行BLOB操作... 目录一、缘由二、办法2.1 基本操作2.2 DBMS_LOB包2.3 字节级操作与RAW数据类型2.

C#原型模式之如何通过克隆对象来优化创建过程

《C#原型模式之如何通过克隆对象来优化创建过程》原型模式是一种创建型设计模式,通过克隆现有对象来创建新对象,避免重复的创建成本和复杂的初始化过程,它适用于对象创建过程复杂、需要大量相似对象或避免重复初... 目录什么是原型模式?原型模式的工作原理C#中如何实现原型模式?1. 定义原型接口2. 实现原型接口3