4、工作队列模式：轮询分发、消息确认、消息持久化、公平分发

工作队列模式，说白了就是「把任务分给多个工人去干」。

我刚开始接触 RabbitMQ 时，觉得这东西不就是个消息队列嘛，能有多复杂？直到我在一个实际项目中，因为没处理好消息确认，导致数据丢失了整整一个晚上……嗯，从那以后我再也不敢小看这几个基础特性了。

4.1 轮询分发：默认的「平均主义」

RabbitMQ 默认的分发策略就是轮询。什么意思呢？就是不管你的消费者能力强不强，每人分一个，轮流来。

举个例子，生产者发了 10 条消息，有两个消费者。那么消费者 A 拿第 1、3、5、7、9 条，消费者 B 拿第 2、4、6、8、10 条。一人一半，公平得很。

代码实现很简单：

// 生产者 - 发送消息
const amqp = require('amqplib');

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

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

    for (let i = 0; i < 10; i++) {
        const msg = `Task ${i}`;
        channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });
        console.log(`Sent: ${msg}`);
    }

    setTimeout(() => connection.close(), 500);
}

sendTasks();

// 消费者 - 接收消息
const amqp = require('amqplib');

async function receiveTask(name) {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queue = 'task_queue';

    await channel.assertQueue(queue, { durable: true });
    channel.consume(queue, (msg) => {
        if (msg !== null) {
            console.log(`${name} received: ${msg.content.toString()}`);
            // 模拟处理时间
            setTimeout(() => {
                channel.ack(msg);
            }, 1000);
        }
    }, { noAck: false });
}

receiveTask('Worker A');
receiveTask('Worker B');

4.2 消息确认：别让任务「死得不明不白」

我在项目中遇到过一个问题：消费者拿到消息后，处理到一半崩溃了。消息没了，任务也没执行。这就是典型的「消息丢失」。

解决方案就是消息确认（ACK）。消费者处理完消息后，主动告诉 RabbitMQ：「我搞定了，你可以删掉这条消息了。」如果消费者挂了，RabbitMQ 会把消息重新投递给其他消费者。

实现方式：

// 消费者 - 手动确认
channel.consume(queue, (msg) => {
    if (msg !== null) {
        try {
            // 处理消息
            console.log(`Processing: ${msg.content.toString()}`);
            // 处理完成后确认
            channel.ack(msg);
        } catch (error) {
            // 处理失败，拒绝消息
            channel.nack(msg, false, true); // 重新入队
        }
    }
}, { noAck: false }); // noAck: false 表示手动确认

4.3 消息持久化：重启也不怕

你想想看，如果 RabbitMQ 服务器突然重启了，队列里的消息还在吗？默认情况下，消息是存在内存里的，重启就没了。

要保证消息不丢，需要做两件事：

1. 队列持久化：声明队列时设置 durable: true
2. 消息持久化：发送消息时设置 persistent: true

// 声明持久化队列
await channel.assertQueue('task_queue', { durable: true });

// 发送持久化消息
channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });

配置项	作用	是否必须
队列 durable: true	队列元数据持久化到磁盘	是
消息 persistent: true	消息内容持久化到磁盘	是
手动 ACK	确保消息被处理后才删除	推荐

4.4 公平分发：能者多劳

轮询分发有个问题：如果消费者 A 处理一条消息要 1 秒，消费者 B 要 5 秒。结果就是 A 闲得发慌，B 忙得要死。

公平分发（Fair Dispatch）就是来解决这个问题的。它告诉 RabbitMQ：「别一次性给消费者太多消息，等它处理完了再给新的。」

实现方式：

// 消费者 - 设置预取数量
channel.prefetch(1); // 每次只取一条消息

channel.consume(queue, (msg) => {
    if (msg !== null) {
        console.log(`Processing: ${msg.content.toString()}`);
        setTimeout(() => {
            channel.ack(msg);
        }, Math.random() * 3000); // 模拟不同处理时间
    }
}, { noAck: false });

prefetch(1) 的意思是：消费者每次只从队列里拿一条消息。处理完并确认后，才能拿下一条。这样处理快的消费者就能多干活，处理慢的消费者就少干活。

4.5 实战建议

我个人习惯在生产环境中这样配置：

• 队列设置 durable: true
• 消息设置 persistent: true
• 消费者设置 noAck: false，手动确认
• 消费者设置 prefetch: 1，实现公平分发
• 处理失败的消息，用 nack 重新入队，但设置重试次数限制

我记得有一次线上事故，就是因为没设置 prefetch，结果某个消费者积压了上千条消息，内存直接爆了。嗯，从那以后，prefetch 就成了我的标配。

说白了，工作队列模式的核心就四个字：可靠、公平。可靠靠的是消息确认和持久化，公平靠的是预取数量控制。把这几个点吃透了，大部分任务队列场景都能搞定。