Matarialize中文社区-RabbitMQ 标准队列

RabbitMQ 标准队列

激流勇进 2024-08-11 19:27:32

230 标准队列专栏

RabbitMQ 标准队列（Classic Queue）是消息传递系统中最基本的队列类型，消息按先进先出（FIFO）的顺序存储和传递。以下是标准队列的常见应用场景和详细说明：

1. 任务队列（Task Queue）

场景描述：在后台处理任务时，如图像处理、视频转码、数据分析等，需要将任务分配给多个工作节点进行处理，避免阻塞主应用。
应用说明：在任务队列中，每个任务都会被推送到 RabbitMQ 的标准队列中，多个消费者从队列中获取任务并处理，任务处理完毕后删除队列中的消息。

举例：

应用：电商网站后台处理订单数据，如生成发货单、发送确认邮件等。这些任务可以异步处理，以提高系统的响应速度。

代码示例（Node.js）：

// 生产者：将任务添加到队列
const amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function(error1, channel) {
        if (error1) {
            throw error1;
        }
        const queue = 'task_queue';
        const msg = 'New Order - Order ID: 12345';

        channel.assertQueue(queue, { durable: true });
        channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });
        console.log(" [x] Sent '%s'", msg);
    });
});

// 消费者：从队列中取出任务并处理
amqp.connect('amqp://localhost', function(error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function(error1, channel) {
        if (error1) {
            throw error1;
        }
        const queue = 'task_queue';

        channel.assertQueue(queue, { durable: true });
        channel.prefetch(1);
        console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);
        channel.consume(queue, function(msg) {
            console.log(" [x] Received '%s'", msg.content.toString());
            // 模拟任务处理时间
            setTimeout(() => {
                console.log(" [x] Done");
                channel.ack(msg);
            }, 1000);
        });
    });
});

2. 异步日志处理

场景描述：在高并发的系统中，记录日志的过程可能会影响系统的性能。通过将日志写入消息队列，再由专门的日志处理服务异步地写入到文件或数据库中，可以减轻主系统的压力。
应用说明：当应用程序需要记录日志时，它将日志消息发送到 RabbitMQ 标准队列中，然后日志服务从队列中取出日志并存储在持久化存储中。

举例：

应用：Web 服务器记录用户行为日志，如访问页面、点击按钮等。日志处理程序从队列中读取这些日志并存储到数据库或文件系统中。

代码示例（Node.js）：

// 生产者：将日志消息添加到队列
const logMessage = 'User clicked on the Home button';
channel.assertQueue('log_queue', { durable: true });
channel.sendToQueue('log_queue', Buffer.from(logMessage), { persistent: true });
console.log(" [x] Sent log: '%s'", logMessage);

// 消费者：从队列中读取日志并存储
channel.consume('log_queue', function(msg) {
    console.log(" [x] Log received: '%s'", msg.content.toString());
    // 模拟日志写入操作
    fs.appendFile('app.log', msg.content.toString() + '\n', function (err) {
        if (err) throw err;
        console.log('Log saved!');
        channel.ack(msg);
    });
});

3. 异步通知系统

场景描述：在电子商务、社交网络等应用中，经常需要向用户发送通知，如订单更新、消息提醒等。为了避免阻塞主应用，可以使用标准队列进行异步通知处理。
应用说明：将通知任务放入 RabbitMQ 的标准队列中，由专门的服务来处理并发送通知，确保主应用的响应时间不受影响。

举例：

应用：电商平台在用户下单成功后发送订单确认短信或邮件。通知服务从队列中取出任务，调用第三方 API 发送短信或邮件。

代码示例（Node.js）：

const notificationMsg = 'Order confirmed: Order ID 12345';
channel.assertQueue('notification_queue', { durable: true });
channel.sendToQueue('notification_queue', Buffer.from(notificationMsg), { persistent: true });
console.log(" [x] Sent notification: '%s'", notificationMsg);

channel.consume('notification_queue', function(msg) {
    console.log(" [x] Notification to send: '%s'", msg.content.toString());
    // 调用第三方API发送通知
    sendNotificationAPI(msg.content.toString(), () => {
        console.log('Notification sent!');
        channel.ack(msg);
    });
});

4. 工作分配（Work Distribution）

场景描述：在分布式系统中，将工作任务分配给多个消费者处理，以充分利用系统资源，实现负载均衡。
应用说明：标准队列允许将任务均匀分配给多个消费者，避免某个消费者过载。

举例：

应用：图片处理应用程序，上传的图片需要经过多个步骤处理（如缩放、裁剪、添加水印），每个步骤可以由不同的消费者处理。

代码示例：

const imageTask = 'Resize Image ID: 54321';
channel.assertQueue('image_task_queue', { durable: true });
channel.sendToQueue('image_task_queue', Buffer.from(imageTask), { persistent: true });
console.log(" [x] Sent image task: '%s'", imageTask);

channel.consume('image_task_queue', function(msg) {
    console.log(" [x] Processing image task: '%s'", msg.content.toString());
    // 模拟图片处理
    processImageTask(msg.content.toString(), () => {
        console.log('Image processed!');
        channel.ack(msg);
    });
});