2

我正在使用 kafka 的 nodejs 中设置服务器。为了做到这一点，我正在使用 kafkaJS。这里的问题是，我不希望每次有人发送帖子或从 kafka 读取值时服务器都会断电。

到目前为止，我已经尝试过这个来读取数据

const readData = async (receiver, stream_name) => {
  return new Promise((resolve, reject) => {
    consumer.connect()
    consumer.subscribe({ topic: stream_name })
      .catch(e => {
        console.log(stream_name)
        console.log('failed to subscribe to topic')
        console.log('err: ', e)
        reject(e)
      })
    consumer.run({
      eachMessage: async ({ topic, message }) => {
        receiver(`- ${topic} ${message.timestamp} ${message.key}#${message.value}`)
        resolve('ok')
      }
    })
      .then(e => {
        console.log('reading')
        resolve('read')
      })
      .catch(e => {
        console.log('messed up', e)
        reject('fail')
      })
    setTimeout(() => {
      console.log('egging')
      return 0
    }, 10000)
  })
}

这是创建数据

const pushData = async payload => {
  return new Promise((resolve, reject) => {
    producer.send(payload)
            .then( e => {
              resolve(e)
            })
            .catch(e => {
              console.log('Error pushing data to kafka broker:\n', e)
              reject(e)
            })
  })
}

// run ({String topic, {String key, JSON value[]} messages} payload)
const putData = async payload => {
  console.log('Connecting to kafka server...')
  await producer.connect()
  const a = await pushData(payload)
  console.log('Data sent: ', a)
  await producer.disconnect()
  process.abort()
}

此代码运行良好，但需要放下它才能退出该方法。我想要一个解决方案，我可以杀死正在执行 kafka 接口的进程/线程或定期退出该方法。

任何帮助表示赞赏。

4

0 回答 0