Matarialize中文社区-RabbitMQ死信队列（Dead Letter Queue, DLQ）的使用详解

RabbitMQ死信队列（Dead Letter Queue, DLQ）的使用详解

激流勇进 2024-10-10 15:43:42

114 死信队列,Dead Letter Queue, DLQ 专栏

什么是死信队列（DLQ）？

在RabbitMQ中，死信队列（Dead Letter Queue, DLQ）是用来处理无法被正常消费的消息的队列。当消息因某种原因无法被消费时，它们会被转移到一个专门的死信队列中，方便后续的处理和监控。

死信的产生原因

消息过期：消息在队列中存在的时间超过了设定的过期时间。
未被消费的消息：消费者在处理消息时未能确认消息（ack），导致消息被重新入队，但依然未能被消费。
消息拒绝：消费者拒绝（reject）消息，并且未设置“重新入队”的选项。

死信队列的工作原理

当一条消息成为“死信”时，RabbitMQ会根据配置的参数将该消息转移到指定的死信队列中。死信队列可以是同一RabbitMQ实例的另一个队列，也可以是其他队列。

死信队列的配置

为了使用死信队列，我们需要在创建正常队列时设置一些参数：

x-dead-letter-exchange：指定死信队列的交换机。
x-dead-letter-routing-key：指定死信队列的路由键。
x-message-ttl：可选，设置消息的过期时间。

使用Node.js实现死信队列

1. 安装与配置RabbitMQ

确保RabbitMQ已经安装并启用管理插件。可以通过访问 http://localhost:15672 来管理RabbitMQ。

2. 创建正常队列与死信队列

首先，我们需要创建一个正常的队列和一个死信队列。以下是相关的代码示例：

安装依赖

确保你的Node.js项目中安装了 amqplib：

npm install amqplib

创建队列的代码示例

const amqp = require('amqplib');

async function createQueues() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const normalQueue = 'normal_queue';
    const deadLetterQueue = 'dead_letter_queue';

    // 创建死信队列
    await channel.assertQueue(deadLetterQueue, {
        durable: true
    });

    // 创建正常队列，指定死信交换机和路由键
    await channel.assertQueue(normalQueue, {
        durable: true,
        arguments: {
            'x-dead-letter-exchange': '', // 使用默认交换机
            'x-dead-letter-routing-key': deadLetterQueue // 指定死信队列
        }
    });

    console.log(`创建正常队列: ${normalQueue}`);
    console.log(`创建死信队列: ${deadLetterQueue}`);

    // 关闭连接
    await channel.close();
    await connection.close();
}

createQueues().catch(console.error);

3. 向正常队列发送消息

接下来，我们可以向正常队列发送消息。以下是发送消息的示例代码：

async function sendMessage() {
    const normalQueue = 'normal_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const message = 'Hello, Normal Queue!';
  
    // 发送消息
    channel.sendToQueue(normalQueue, Buffer.from(message), { persistent: true });
    console.log(`发送消息: ${message}`);

    // 关闭连接
    await channel.close();
    await connection.close();
}

sendMessage().catch(console.error);

4. 从正常队列接收消息

接收消息的代码示例如下。为了模拟消息消费失败，我们在处理过程中拒绝消息。

async function receiveMessage() {
    const normalQueue = 'normal_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    await channel.assertQueue(normalQueue, { durable: true });

    console.log(`等待接收消息来自 ${normalQueue}...`);

    channel.consume(normalQueue, (msg) => {
        if (msg !== null) {
            console.log(`接收到消息: ${msg.content.toString()}`);
            // 拒绝消息并不重新入队
            channel.reject(msg, false); // false表示不重新入队
            console.log(`消息被拒绝并转移到死信队列`);
        }
    }, { noAck: false });
}

receiveMessage().catch(console.error);

5. 从死信队列接收消息

最后，我们需要从死信队列中接收被转移的消息。代码示例如下：

async function receiveDeadLetter() {
    const deadLetterQueue = 'dead_letter_queue';
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    await channel.assertQueue(deadLetterQueue, { durable: true });

    console.log(`等待接收消息来自 ${deadLetterQueue}...`);

    channel.consume(deadLetterQueue, (msg) => {
        if (msg !== null) {
            console.log(`接收到死信消息: ${msg.content.toString()}`);
            // 手动确认消息
            channel.ack(msg);
        }
    }, { noAck: false });
}

receiveDeadLetter().catch(console.error);