kafka-node 异步消费者处理程序答案

【问题标题】：kafka-node asynchronous consumer handlerkafka-node 异步消费者处理程序
【发布时间】：2019-01-23 10:45:20
【问题描述】：

这就是我的消费者的初始化方式：

const client = new kafka.Client(config.ZK_HOST)
const consumer = new kafka.Consumer(client, [{ topic: config.KAFKA_TOPIC, offset: 0}],
{
    autoCommit: false
})

现在消费者consumer.on('message', message => applyMessage(message))

问题是applyMessage 使用knex 与数据库对话，代码如下所示：

async function applyMessage(message: kafka.Message) {
    const usersCount = await db('users').count()
    // just assume we ABSOLUTELY need to calculate a number of users,
    // so we need previous state
    await db('users').insert(inferUserFromMessage(message))
}

上面的代码使applyMessage 对kafka 中的所有消息并行执行，所以在上面的代码中，假设数据库中还没有用户，usersCount 将始终为 0，即使对于来自的第二条消息自从第一次调用 applyMessage 插入用户以来，它应该是 1 的 kafka。

如何以所有applyMessage 函数按顺序运行的方式“同步”代码？

【问题讨论】：

标签： node.js apache-kafka kafka-consumer-api

【解决方案1】：

您需要实现某种互斥锁。基本上是一个将事情排队以同步执行的类。示例

var Mutex = function() {
  this.queue = [];
  this.locked = false;
};

Mutex.prototype.enqueue = function(task) {
  this.queue.push(task);
  if (!this.locked) {
    this.dequeue();
  }
};

Mutex.prototype.dequeue = function() {
  this.locked = true;
  const task = this.queue.shift();
  if (task) {
    this.execute(task);
  } else {
    this.locked = false;
  }
};

Mutex.prototype.execute = async function(task) {
  try { await task(); } catch (err) { }
  this.dequeue();
}

为了使其正常工作，您的 applyMessage 函数（无论处理 Kafka 消息）需要返回一个 Promise - 还要注意 async 已从父函数移动到返回的 Promise 函数：

function applyMessage(message: kafka.Message) {
  return new Promise(async function(resolve,reject) {
    try {
      const usersCount = await db('users').count()
      // just assume we ABSOLUTELY need to calculate a number of users,
      // so we need previous state
      await db('users').insert(inferUserFromMessage(message))
      resolve();
    } catch (err) {
      reject(err);
    }
  });
}

最后，applyMessage 的每次调用都需要加入 Mutex 队列，而不是直接调用：

var mutex = new Mutex();
consumer.on('message', message => mutex.enqueue(function() { return applyMessage(message); }))

【讨论】：