Node&RabbitMQ系列二 延迟｜死信队列

// producer.ts
import * as amqp from 'amqplib'

const url = `amqp://localhost:5672`
async function publish(msg: string, ttl: number) {
    const exchange = 'my-delayed-exchange';
    const exchangeType = 'x-delayed-message'; // x-delayed-message 交换机的类型
    const routingKey = 'my-delayed-routingKey';

    const connect = await amqp.connect(url);
    const channel = await connect.createChannel();
    await channel.assertExchange(exchange, exchangeType, { durable: true, arguments: {'x-delayed-type': 'direct' }})
    console.log('发布消息', msg, ttl, routingKey);
    channel.publish(exchange, routingKey, Buffer.from(msg), {
        headers: {
            'x-delay': ttl, // 一定要设置，否则无效
        }
    });
    channel.close();
}

(async function test(){
    await publish('msg0 1S Expire', 1000);
    await publish('msg0 2S Expire', 2000);
    await publish('msg0 3S Expire', 3000);
    // 最后一个不会触发
    process.exit(0);
})();

// consumer.ts
import * as amqp from 'amqplib'

const url = `amqp://localhost:5672`;

(async function publish() {
    const exchange = 'my-delayed-exchange';
    const exchangeType = 'x-delayed-message';
    const routingKey = 'my-delayed-routingKey';
    const queueName = 'my-delayed-queue';
    try {
        const connect = await amqp.connect(url);
        const channel = await connect.createChannel();
        await channel.assertExchange(exchange, exchangeType, { durable: true, arguments: { 'x-delayed-type': 'direct' } })
        const queueA = await channel.assertQueue(queueName);
        console.log(queueA);

        await channel.bindQueue(queueA.queue, exchange, routingKey);
        await channel.consume(queueA.queue, msg => {
            console.log("接受到的消息", msg.content.toString());
        }, { noAck: true });
    } catch (error) {
        console.log(error);
    }
})();

// producerDLX.ts
import * as amqp from 'amqplib'
/** 
 * 1. 正常发送消息到交换机
 * 2. 如果超时，发送到死信队列
 * 3. 对应 consumerDLX.ts 、consumerEX.ts
 * 4. consumerEX.ts 消费生产的消息
 * 5. consumerDLX.ts 消费 生产的消息 被会收到 死信队列的消息
*/

(async function publish() {

    const url = `amqp://localhost:5672`;
    // 默认交换机-队列-路由
    const exchange = 'ex.exchange';
    const routingKey = 'ex.routerkey';

    const connect = await amqp.connect(url);
    const channel = await connect.createChannel();
    // 默认交换机
    await channel.assertExchange(exchange, 'direct', { durable: false });
    // 发布消息
    channel.publish(exchange, routingKey, Buffer.from('hello world'), {
        expiration: 3000
    });
    await sleep(1);
    await connect.close();
    process.exit(0);
})();


function sleep(time: number) {
    return new Promise((resolve) => setTimeout(resolve, time*1000));
}

import * as amqp from 'amqplib'

/**
 * 对应producerDLX.ts 发送的消息进行消费
*/

(async function consumer() {

    // 交换机-路由
    const url = `amqp://localhost:5672`;
    // 死信交换机-路由
    const deadLetterExchange = 'dlx.exchange';
    const deadLetterRoutingKey = 'dlx.routingkey' 
    // 默认交换机-队列-路由
    const exchange = 'ex.exchange';
    const queueExName = 'ex.queue';
    const routingKey = 'ex.routerkey';

    const connect = await amqp.connect(url);
    const channel = await connect.createChannel();
    await channel.assertExchange(exchange, 'direct', { durable: false });
    const queueEX = await channel.assertQueue(queueExName, {
        exclusive: false,
        deadLetterExchange,
        deadLetterRoutingKey,
    });
    await channel.bindQueue(queueExName, exchange, routingKey);
    await channel.consume(queueEX.queue, msg => {
        console.log("消费队列", msg);
    }, {noAck: true});
})();

import * as amqp from 'amqplib'
/**
 * 针对死信队列，进行消费
 * 对应producerDLX.ts
*/

(async function consumer() {

    const url = `amqp://localhost:5672`;
    // 死信交换机-路由
    const deadLetterExchange = 'dlx.exchange';
    const deadLetterRoutingKey = 'dlx.routingkey'
    const deadLetterQueue = 'dlx.queue';

    const connect = await amqp.connect(url);
    const channel = await connect.createChannel();
    // 默认交换机
    await channel.assertExchange(deadLetterExchange, 'direct', { durable: false });
    // 队列，超时发动到死信队列
    const queueDLX = await channel.assertQueue(deadLetterQueue, {exclusive: false});
    await channel.bindQueue(deadLetterQueue, deadLetterExchange, deadLetterRoutingKey);
    await channel.consume(queueDLX.queue, msg => {
        console.log("消费死信队列", msg);
    }, {noAck: true});
})();