如何确保BullMQ队列将数据推送到Redis集群指定节点？

我们已在Node.js应用中集成BullMQ，并采用Redis集群作为生产环境。根据BullMQ与Redis集群文档，创建队列时可设置键前缀，内部会对键进行CRC16运算后取模16384，但这种方式无法控制数据的目标节点，若两个队列前缀的哈希值落在同一节点，可能引发热点分片问题。请问是否有办法确保BullMQ中的每个队列都将数据推送到Redis集群内的预定义节点？

示例代码：

const Redis = require('ioredis');
const { Queue, Worker } = require('bullmq');

const clusterNodes = [
  { port: 6380, host: "127.0.0.1" },
  { port: 6381, host: "127.0.0.1" },
  { port: 6382, host: "127.0.0.1" }
];

const cluster = new Redis.Cluster(clusterNodes);

async function bullMqTest() {

  const queue1 = new Queue(`myQueue1`, {
    connection: cluster,
    prefix: "{queue_1}",
  });
  const queue2 = new Queue(`myQueue2`, {
    connection: cluster,
    prefix: "{queue_2}",
  });
  const queue3 = new Queue(`myQueue3`, {
    connection: cluster,
    prefix: "{queue_3}",
  });
  const queue4 = new Queue(`myQueue4`, {
    connection: cluster,
    prefix: "{queue_4}",
  });
  const queue5 = new Queue(`myQueue5`, {
    connection: cluster,
    prefix: "{queue_5}",
  });
  const queue6 = new Queue(`myQueue6`, {
    connection: cluster,
    prefix: "{queue_6}",
  });


  queue1.add('myJob', { foo: 'bar' });
  queue2.add('myJob', { foo: 'baz' });
  queue3.add('myJob', { foo: 'qux' });
  queue4.add('myJob', { foo: 'xyz' });
  queue5.add('myJob', { foo: 'lmn' });
  queue6.add('myJob', { foo: 'qpr' });
}

module.exports = { pushDataToRedis, fetchDataFromRedis, bullMqTest };

要实现每个BullMQ队列绑定到Redis集群的指定节点，核心思路是利用Redis集群的哈希标签规则手动匹配目标节点哈希槽，或者为每个队列创建带路由偏好的独立Redis连接，以下是具体实现方案：

方法1：手动计算哈希标签匹配目标节点槽位 #

Redis集群通过CRC16(key) % 16384确定键对应的哈希槽，而哈希标签{xxx}会让Redis只计算标签内字符串的哈希值。通过以下步骤实现队列与节点的绑定：

• 执行redis-cli -c -p 集群端口 CLUSTER SLOTS命令，获取各节点负责的哈希槽范围（例如节点6380负责0-5460，6381负责5461-10922）。
• 使用CRC16工具计算字符串的哈希值，找到能落在目标节点槽范围内的字符串，将其作为队列前缀的标签内容（格式为{自定义标签}）。
• 修改队列初始化代码：

// 假设已计算好对应各节点的标签
const queue1 = new Queue(`myQueue1`, {
  connection: cluster,
  prefix: "{slot_0}", // 哈希值落在6380节点的槽范围
});
const queue2 = new Queue(`myQueue2`, {
  connection: cluster,
  prefix: "{slot_5461}", // 哈希值落在6381节点的槽范围
});

方法2：为每个队列创建带路由偏好的独立连接 #

利用ioredis的集群配置，为每个队列创建独立的连接并强制路由到目标节点，示例代码如下：

const Redis = require('ioredis');
const { Queue, Worker } = require('bullmq');

// 创建指定目标节点的集群连接
function createClusterConnection(targetPort) {
  return new Redis.Cluster(
    [
      { port: 6380, host: "127.0.0.1" },
      { port: 6381, host: "127.0.0.1" },
      { port: 6382, host: "127.0.0.1" }
    ],
    {
      // 手动指定槽映射，确保队列键落在目标节点的槽范围
      slots: [
        [0, 5460, ["127.0.0.1", targetPort]],
        [5461, 10922, ["127.0.0.1", targetPort]],
        [10923, 16383, ["127.0.0.1", targetPort]]
      ]
    }
  );
}

async function bullMqTest() {
  // 队列1绑定到6380节点
  const queue1 = new Queue(`myQueue1`, {
    connection: createClusterConnection(6380),
    prefix: "{queue_1}",
  });
  // 队列2绑定到6381节点
  const queue2 = new Queue(`myQueue2`, {
    connection: createClusterConnection(6381),
    prefix: "{queue_2}",
  });
  // 队列3绑定到6382节点
  const queue3 = new Queue(`myQueue3`, {
    connection: createClusterConnection(6382),
    prefix: "{queue_3}",
  });

  queue1.add('myJob', { foo: 'bar' });
  queue2.add('myJob', { foo: 'baz' });
  queue3.add('myJob', { foo: 'qux' });
}

module.exports = { pushDataToRedis, fetchDataFromRedis, bullMqTest };

注意事项 #

• 方法1需要维护哈希标签与槽位的对应关系，集群拓扑变更（如增减节点）时需重新计算标签。
• 方法2的独立连接会增加资源消耗，需根据队列数量和业务负载评估合理性。
• Worker必须使用与对应队列相同的连接配置，否则会出现无法找到任务的问题。

内容的提问来源于stack exchange，提问作者Ankit Kadam