Redis消息队列场景下如何实现单任务单Worker的负载均衡？

阿华AIGC实验室

2026-5-12

解决PM2多Worker下Redis消息重复消费的问题

你踩的这个坑其实很常见——当多个独立的Worker进程同时监听同一个Redis队列时，如果没有用正确的消息分发机制，就会出现同一条任务被多个Worker重复执行的情况。核心原因是Redis的基础LIST结构本身没有内置的"任务独占分配"能力，得靠我们在消费端或者队列层做处理。下面给你几个实用的解决方案，按推荐优先级排序：

1. 使用Redis Stream + 消费者组（最推荐的现代方案）

Redis 5.0以上引入的Stream类型专为消息队列设计，原生支持**消费者组（Consumer Group）**机制，能自动把任务均匀分配给组内的各个消费者（也就是你的Worker进程），完全避免重复消费。

具体怎么做：

生产者端：用XADD把任务消息写入Stream
消费者端：每个Worker加入同一个消费者组，用XREADGROUP以阻塞方式读取消息，Redis会确保同一条消息只分给组里的一个Worker
处理完成后调用XACK确认消息，Redis就会把这条消息从待处理列表中移除

示例代码片段：

// Worker 端代码
const redis = require('redis');
const client = redis.createClient();

// 初始化消费者组（只需要执行一次，或者在Worker启动时判断是否存在）
client.xgroupCreate('task_stream', 'worker_group', '$', { MKSTREAM: true }, (err) => {
  if (err && err.message !== 'BUSYGROUP Consumer Group name already exists') {
    console.error('创建消费者组失败:', err);
  }
});

// 持续读取消息
function consumeTasks() {
  client.xreadgroup('GROUP', 'worker_group', `worker_${process.pid}`, 'BLOCK', 0, 'COUNT', 1, 'STREAMS', 'task_stream', '>', (err, replies) => {
    if (err) {
      console.error('读取消息失败:', err);
      return setTimeout(consumeTasks, 1000);
    }
    if (replies) {
      const [stream, messages] = replies[0];
      const [messageId, taskData] = messages[0];
      // 处理任务
      console.log(`Worker ${process.pid} 处理任务:`, taskData);
      // 任务完成后确认
      client.xack('task_stream', 'worker_group', messageId, (err) => {
        if (err) console.error('确认消息失败:', err);
        consumeTasks();
      });
    } else {
      consumeTasks();
    }
  });
}

consumeTasks();

用PM2启动多实例时，每个Worker会以不同的消费者ID（比如带pid的worker_1234）加入组，Redis会自动把任务轮询分配给各个Worker，完全不用担心重复消费。

2. 使用原子性的消息转移 + ACK机制（兼容旧版Redis）

如果你的Redis版本低于5.0，可以用RPOPLPUSH（或者BRPOPLPUSH阻塞版本）实现"任务锁定"：

把任务从主队列移到一个"处理中"的临时队列
只有成功完成这个原子操作的Worker才能拿到任务
处理完成后，从临时队列删除该任务；如果处理失败，再把任务移回主队列

示例代码：

const redis = require('redis');
const client = redis.createClient();

function consumeTasks() {
  // 阻塞式从主队列取消息，同时移到处理中队列
  client.brpoplpush('task_queue', 'processing_queue', 0, (err, taskData) => {
    if (err) {
      console.error('获取任务失败:', err);
      return setTimeout(consumeTasks, 1000);
    }
    // 处理任务
    console.log(`Worker ${process.pid} 处理任务:`, taskData);
    // 处理完成，从处理中队列删除
    client.lrem('processing_queue', 1, taskData, (err) => {
      if (err) console.error('清理处理中任务失败:', err);
      consumeTasks();
    });
  });
}

consumeTasks();

这个方案的核心是BRPOPLPUSH是原子操作——同一时刻只有一个Worker能把消息从主队列移到处理队列，其他Worker拿不到这条消息，自然不会重复处理。

3. 使用现成的任务队列库（最省心的方案）

如果不想自己实现队列的负载均衡和可靠性机制，直接用成熟的Node.js任务队列库，比如BullMQ或者Bee-Queue。这些库已经封装好了Redis的消息分发、负载均衡、重试、失败重试等功能，多Worker场景下开箱即用。

以BullMQ为例，代码非常简洁：

// Worker 端代码
const { Worker } = require('bullmq');

const worker = new Worker('task_queue', async (job) => {
  // 处理任务
  console.log(`Worker ${process.pid} 处理任务:`, job.data);
  // 返回处理结果
  return { result: 'success' };
});

worker.on('completed', (job) => {
  console.log(`任务 ${job.id} 处理完成`);
});

worker.on('failed', (job, err) => {
  console.log(`任务 ${job.id} 处理失败:`, err);
});

当你用pm2 start -i max myWorker.js启动多实例时，BullMQ会自动在多个Worker之间做负载均衡，每个任务只会被分配给一个Worker处理，完全不用你关心底层的Redis操作。