typescript 实现RabbitMQ死信队列和延迟队列(订单10分钟未付归还库存)的过程

sudo pacman -S rabbitmq rabbitmqadmin

sudo rabbitmq-plugins enable rabbitmq_management
sudo rabbitmq-server

rabbitmqctl add_user mingcai password

rabbitmqctl set_permissions -p / mingcai ".*" ".*" ".*"

rabbitmqctl set_user_tags mingcai administrator

// 引入amqplib库
import * as amqp from 'amqplib';
// 连接到RabbitMQ服务器
const connection = await amqp.connect('amqp://localhost');
// 创建Channel
const channel = await connection.createChannel();
// 定义队列
const queueName = 'my_queue';
await channel.assertQueue(queueName, {
  // 设置死信交换机
  deadLetterExchange: 'my_dead_letter_exchange'
});
// 消费消息
channel.consume(queueName, (msg) => {
  // 处理消息
  if (msg) {
    // 处理失败，拒绝消息并将其重新放回队列
    // channel.reject(msg, true); // 第二个参数设为 true 表示将消息重新放回队列
    // 处理失败，拒绝消息
    channel.reject(msg, false); // 第二个参数设为 false 表示将消息投递到死信队列
     // or 处理失败，拒绝消息并将其重新放回死信队列
    channel.nack(msg, false, false); // 第二个参数设为 false 表示不将消息重新放回原队列，第三个参数设为 false 表示不拒绝当前和之前所有未确认的消息
  }
});

// 发布消息
await channel.sendToQueue(queueName, Buffer.from('Hello'), {
  expiration: '60000' // 设置过期时间为60秒
});

// 定义队列
await channel.assertQueue(queueName, {
  maxLength: 100, // 设置最大队列长度为100
  deadLetterExchange: 'my_dead_letter_exchange'
});

npm install amqplib --save
npm install @types/amqplib --save-dev

export RABBITMQ_SERVER=127.0.0.1
export RABBITMQ_PORT=5672   
export RABBITMQ_USER=mingcai     
export RABBITMQ_PASSWORD=password
rabbitmqadmin list exchanges

rabbitmqadmin status

rabbitmqadmin list exchanges

rabbitmqadmin list queues

rabbitmqadmin declare exchange name=my_exchange type=direct

rabbitmqadmin declare queue name=my_queue

rabbitmqadmin declare binding source=my_exchange destination=my_queue routing_key=my_routing_key

rabbitmqadmin publish exchange=my_exchange routing_key=my_routing_key payload="Hello, RabbitMQ!"

rabbitmqadmin get queue=my_queue

// delayProducer.ts
import * as amqp from 'amqplib';
async function setupRouting() {
    const connection = await amqp.connect('amqp://mingcai:password@127.0.0.1');
    const channel = await connection.createChannel();
    const exchange = 'routing_exchange';
    // 定义 dlx-exchange
    const dlxExchangeName = 'dlx-exchange';
    // 声明交换机
    await channel.assertExchange(exchange, 'direct', {durable: true});
    await channel.assertExchange(dlxExchangeName, 'direct', { durable: true });//消息防止丢失
    const dlxqueueBindings= [
        {
            dlxQueueName: 'dlx-3_second_queue', routingKey: 'fast',
        },
        {
            dlxQueueName: 'dlx-8_second_queue', routingKey: 'slow'
        }
    ];
    for (const binding of dlxqueueBindings) {
        // 绑定延迟死信队列
        await channel.assertQueue(binding.dlxQueueName );
        //死信交换机和死信队列绑定 Routing key fast 的消息
        await channel.bindQueue(binding.dlxQueueName, dlxExchangeName, binding.routingKey); // 将 dlx-queue 绑定到死信交换机
    }
    // 定义队列和路由键的映射
    const queueBindings = [
        {
            queue: '3_second_queue', routingKey: 'fast', arguments: {
                'x-message-ttl': 3000, // TTL 设置为 3 秒 消息被拒绝或过期时将重新发布到的交换器的可选名称。
                'x-dead-letter-exchange': 'dlx-exchange'//消息被拒绝或过期时将重新发布到的交换器的可选名称
            }
        },
        {
            queue: '8_second_queue', routingKey: 'slow', arguments: {
                'x-message-ttl': 8000, // TTL 设置为 8 秒 消息被拒绝或过期时将重新发布到的交换器的可选名称。
                'x-dead-letter-exchange': 'dlx-exchange'//消息被拒绝或过期时将重新发布到的交换器的可选名称
            }
        }
    ];
    // 声明队列，并将队列绑定到交换机上
    for (const binding of queueBindings) {
        await channel.assertQueue(binding.queue, {durable: true, arguments: binding.arguments});
        await channel.bindQueue(binding.queue, exchange, binding.routingKey);
    }
    for (let i = 0; i < 10; i++) {
        await new Promise((resolve) => {
            setTimeout(() => {
                resolve(1)
            }, 1000)
        })
        const chinaTime = new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
        console.log('当前中国时间：', chinaTime);
        // 发送消息到交换机，并设置不同的路由键
        await sendMessage(channel, exchange, 'fast', `[${i}] ${chinaTime} Message for the fast queue`);
        await sendMessage(channel, exchange, 'slow', `[${i}] ${chinaTime} Message for the slow queue`);
    }
    // 关闭连接
    setTimeout(async () => {
        await channel.close();
        await connection.close();
    }, 10000); // 在 10 秒后关闭连接
}
async function sendMessage(channel: amqp.Channel, exchange: string, routingKey: string, message: string) {
    channel.publish(exchange, routingKey, Buffer.from(message));
    console.log(`Sent message '${message}' with routing key '${routingKey}'`);
}
setupRouting().catch(console.error);

//消费者 dlx-3_second_queue.ts
import * as amqp from 'amqplib';
async function setupRouting() {
    const connection = await amqp.connect('amqp://mingcai:password@127.0.0.1');
    const channel = await connection.createChannel();
    let queue = 'dlx-3_second_queue'
    // 定义队列和路由键的映射
    await channel.consume(queue, (msg) => {
        if (msg !== null) {
            const chinaTime = new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
            console.log(`Received message ${chinaTime}'${msg.content.toString()}' from queue '${queue}'`);
            channel.ack(msg); // 确认消息已被处理
        }
    });
}
setupRouting().catch(console.error);
//dlx-8_second_queue.ts
import * as amqp from 'amqplib';
async function setupRouting() {
    const connection = await amqp.connect('amqp://mingcai:password@127.0.0.1');
    const channel = await connection.createChannel();
    let queue = 'dlx-8_second_queue'
    // 定义队列和路由键的映射
    await channel.consume(queue, (msg) => {
        if (msg !== null) {
            const chinaTime = new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
            console.log(`Received message ${chinaTime}'${msg.content.toString()}' from queue '${queue}'`);
            channel.ack(msg); // 确认消息已被处理
        }
    });
}
setupRouting().catch(console.error);