node.js - 从单个 REDIS 实例读取的 Nodejs 集群架构

Question

我正在使用 Nodejs cluster 模块让多个 worker 运行。我创建了一个基本架构，其中将有一个 MASTER 进程，它基本上是一个处理多个请求的快速服务器，MASTER 的主要任务是将来自请求的传入数据写入 REDIS 实例。其他 worker (numOfCPUs - 1)将是非主人，即他们不会处理任何请求，因为他们只是消费者。我有两个功能，即 ABC 和 DEF。我通过分配类型将非 master worker 平均分布在各个功能中。

例如:在 8 核机器上:

1 将是通过 express 服务器处理请求的 MASTER 实例

剩余的 (8 - 1 = 7) 将平均分配。 4 个特征:ABD 和 3 个特征:DEF。

非 master worker 基本上是消费者，即他们从 REDIS 中读取数据，其中只有 MASTER worker 可以写入数据。

下面是相同的代码:

if (cluster.isMaster) {
  // Fork workers.
  for (let i = 0; i < numCPUs - 1; i++) {
    ClusteringUtil.forkNewClusterWithAutoTypeBalancing();
  }

  cluster.on('exit', function(worker) {
    console.log(`Worker ${worker.process.pid}::type(${worker.type}) died`);
    ClusteringUtil.removeWorkerFromList(worker.type);
    ClusteringUtil.forkNewClusterWithAutoTypeBalancing();
  });

  // Start consuming on server-start
  ABCConsumer.start();
  DEFConsumer.start();

  console.log(`Master running with process-id: ${process.pid}`);
} else {
  console.log('CLUSTER  type', cluster.worker.process.env.type, 'running on', process.pid);
  if (
    cluster.worker.process.env &&
    cluster.worker.process.env.type &&
    cluster.worker.process.env.type === ServerTypeEnum.EXPRESS
  ) {
    // worker for handling requests
    app.use(express.json());
    ...
  }
{

除了消费者从 REDIS 读取数据外，一切正常。由于特定功能有多个消费者，每个消费者都读取相同的消息并开始单独处理，这是我不希望的。如果有 4 个消费者，其中 1 个被标记为繁忙并且在有空之前不能消费，则有 3 个可用。一旦 MASTER 在 REDIS 中写入该特定功能的消息，问题是该功能的所有 3 个可用消费者都开始使用。这意味着对于单个消息，工作是根据可用消费者的数量完成的。

const stringifedData = JSON.stringify(req.body);
  const key = uuidv1();

  const asyncHsetRes = await asyncHset(type, key, stringifedData);

  if (asyncHsetRes) {
    await asyncRpush(FeatureKeyEnum.REDIS.ABC_MESSAGE_QUEUE, key);
    res.send({ status: 'success', message: 'Added to processing queue' });
  } else {
    res.send({ error: 'failure', message: 'Something went wrong in adding to queue' });
  }

消费者简单地接受消息并在忙碌时停止

module.exports.startHeartbeat = startHeartbeat = async function(config = {}) {
  if (!config || !config.type || !config.listKey) {
    return;
  }

  heartbeatIntervalObj[config.type] = setInterval(async () => {
    await asyncLindex(config.listKey, -1).then(async res => {
      if (res) {
        await getFreeWorkerAndDoJob(res, config);
        stopHeartbeat(config);
      }
    });
  }, HEARTBEAT_INTERVAL);
};

理想情况下，一条消息应该只被该特定功能的一个消费者阅读。消费后，它被标记为忙碌，所以它不会进一步消费，直到空闲(我已经处理过这个)。下一条消息只能由其他可用消费者中的一个消费者处理。

请帮我解决这个问题。同样，我希望只有一个免费消费者可以阅读一条消息，而其他免费消费者应该等待新消息。

谢谢

Answer 1

我不确定我是否完全理解您的 Redis 消费者架构，但我觉得它与 Redis 本身的用例相矛盾。您要实现的本质上是一种基于队列的消息传递，能够在消息完成后提交消息。

Redis 有自己的发布/订阅功能，但它是建立在火后遗忘原则之上的。它不区分消费者 - 它只是将数据发送给所有消费者，假设它们的逻辑是处理传入数据。

我建议您使用 RabbitMQ 等队列服务器。您可以使用 AMQP 0-9-1 支持的一些功能来实现您的目标:消息确认、消费者的预取计数等。您可以使用非常灵活的配置来设置您的集群，例如 ok，我想要 X 个消费者，并且每个消费者一次可以处理 1 个唯一的(!)消息，并且只有在他们让服务器(rabbitmq ) 知道他们已成功完成消息处理。这是高度可配置和健壮的。

但是，如果您想通过一些完全托管的服务实现无服务器，这样您就不会像虚拟机或其他任何东西一样配置来运行您选择的消息队列服务器，您可以使用 AWS SQS。它具有非常相似的 API 和功能列表。

希望对您有所帮助!