kafka-node асинхронный потребительский обработчик

Question

kafka-node асинхронный потребительский обработчик

для моих проектов Я использую свою библиотеку commons-version https://github.com/raydac/commons-version , она содержит два вспомогательных класса - для разбора версии (анализируемая версия может быть сравнена с другой объект версии, потому что он сопоставим) и VersionValidator, который позволяет проверять версию для некоторого выражения, например !=ide-1.1.1,>idea-1.3.4-SNAPSHOT;<1.2.3

1

node.js apache-kafka kafka-consumer-api

задан Nemoden 18 January 2019 в 05:33

1 ответ

Другие вопросы по тегам:

node.js apache-kafka kafka-consumer-api

Похожие вопросы:

score 0 · Answer 1

Вам нужно будет внедрить какой-нибудь Mutex. В основном класс, который ставит в очередь вещи для синхронного выполнения. Пример

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 - обратите внимание, что асинхронность переместилась из родительской функции в возвращенное Promise. function:

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