时间戳

首页 > 解决方案 > 从单个 REDIS 实例读取 Nodejs 集群架构

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

问题描述

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

例如：在 8 核机器上：

1 将是 MASTER 实例通过快速服务器处理请求

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

非主工作者基本上是消费者，即他们从只有主工作者可以写入数据的 REDIS 中读取。

这是相同的代码：

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);
};

理想情况下，一条消息应该只由该特定功能的一个消费者阅读。消费后，它被标记为忙碌，因此在空闲之前不会进一步消费（我已经处理了这个）。下一条消息只能由其他可用消费者中的一个消费者处理。

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

谢谢

标签： node.jsrediscluster-computing

解决方案

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

Redis 有自己的 pub/sub 功能，但它是建立在 Fire and Forget 原则之上的。它不区分消费者 - 它只是将数据发送给所有消费者，假设其逻辑是处理传入数据。

我推荐你使用像 RabbitMQ 这样的队列服务器。您可以通过 AMQP 0-9-1 支持的一些特性来实现您的目标：消息确认、消费者的预取计数等。您可以使用非常灵活的配置来设置集群，比如 好的，我想要 X 个消费者，每个消费者一次可以处理 1 个唯一（！）消息，只有在他们让服务器（rabbitmq）知道之后，他们才会收到新消息他们成功地完成了消息处理。这是高度可配置和健壮的。

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

希望能帮助到你！

推荐阅读