javascript - 使用 kafka-node 创建的消费者和生产者无法连接到 kafka 的工作实例

Question

我尝试通过 kafka-node npm 库使用消费者和生产者，以便通过 nodejs 使用 kafka 作为消息服务。

问题是，即使我的生产者有时工作，消费者总是给我超时错误，或者它只是在无限循环中释放尝试连接到 kafka，即使 kafka 工作正常。

当 Kafka 在远程 centos7 机器上时，我在我的 windows 机器上使用 kafka-node。即使我将所有代码(消费者和生产者)与 kafka 放在同一台机器上(认为 mabye windows 是问题的一部分)，奇怪的行为仍然存在。

我尝试在生产者控制台内置的 kafka 中发送消息，但我的消费者似乎仍然没有响应主题并收到消息。

这是我的简单生产者代码:

const kclient = new kafka.KafkaClient({kafkaHost:'ADDR:9092'});
kclient.on('error',(err) => {
    console.log(err)
})
kprod = new producer (kclient);
// kconsumer = new consumer(kclient);
kprod.on('error',(err) => {
    console.log(`error: ${err}`);
})

kprod.on('ready',() => {
    console.log(`connected to kafka`);
    let tranNumSentToKafka = 0
    for (let index = 0; index < transArray.length; index++) {
        const element = JSON.stringify(transArray[index])
        console.log(`sending data to kafka`);
        kprod.send([{
            topic:'test',
            messages:element
        }],
            (err,data) => { 
            if(err){console.error(err)}
            else{
                tranNumSentToKafka += 1
                console.log(`data sent: ${JSON.stringify(data)}`);
                console.log(`sent ${tranNumSentToKafka} transactions to kafka`);
            }
        })
    }
})

当我运行它时，有时会出现超时错误，如下所示:

{ TimeoutError: Request timed out after 30000ms
    at new TimeoutError (C:\Users\Yishai Nachaliel\Documents\try\kafka-node-elastic\node_modules\kafka-node\lib\errors\TimeoutError.js:6:9)
    at Timeout.timeoutId._createTimeout [as _onTimeout] (C:\Users\Yishai Nachaliel\Documents\try\kafka-node-elastic\node_modules\kafka-node\lib\kafkaClient.js:1007:14)
    at ontimeout (timers.js:436:11)
    at tryOnTimeout (timers.js:300:5)
    at listOnTimeout (timers.js:263:5)
    at Timer.processTimers (timers.js:223:10) message: 'Request timed out
after 30000ms' }

但有时它会起作用，它会给我以下输出:

...
sent 96 transactions to kafka
data sent: {"test":{"0":15740}}
sent 97 transactions to kafka
data sent: {"test":{"0":15741}}
sent 98 transactions to kafka
data sent: {"test":{"0":15742}}
sent 99 transactions to kafka
data sent: {"test":{"0":15743}}
sent 100 transactions to kafka

这是我的简单消费者:

kafka = require('kafka-node'),
producer = kafka.Producer,
consumer = kafka.Consumer;

const kclient = new kafka.KafkaClient({
    kafkaHost:'10.0.0.55:9092'
    // kafkaHost:'35.186.191.135:9092'
});
kclient.on('ready',() => {
    console.log(`kclient ready`);
    kconsumer = new consumer(kclient,[{
        topic:'test',
        partition:0
    }]);
    kconsumer.on('error',(err) => {
        console.error(` in kconsumer: \n${err}\n`)
    })
    kconsumer.on('ready',() => {
        console.log(`kconsumer ready`);
        kconsumer.on('message',(msg) => {
            console.log(`recived msg: ${msg}`);
        })

    })

})
kclient.on('error',(err) => {
    console.error(`err in kclient: \n${err}\n`)
})

当我在我的 windows 机器上运行它时，我得到:

kclient ready
 in kconsumer:
TimeoutError: Request timed out after 30000ms

当我在centos机器上运行消费者时，我没有得到任何错误，只是卡住了:

kclient ready

永远不会“kconsumer ready”。

在 DEBUG 模式下运行两者只显示:

...
  kafka-node:KafkaClient kafka-node-client reconnecting to ADDR:9092 +1s
  kafka-node:KafkaClient kafka-node-client createBroker ADDR:9092 +2ms
  kafka-node:Consumer connection closed +1s
  kafka-node:KafkaClient kafka-node-client socket closed ADDR:9092 (hadError: true) +3ms
...

当我测试 kafka 和 zookeeper 在我的 centos 机器上是否一切正常时，我使用生产者/消费者控制台对其进行了测试。

另外，我检查了 zookeeper 和 kafka 的日志，没有错误，但没有迹象表明任何生产者或消费者已连接，或发送任何以太消息。

有没有人遇到过kafka-node库的这个问题？

有没有人找到解决方案？

Answer 1

我不确定它是否符合确切的情况，但我有类似的问题行为。

要使用 Kafka，您的客户端应该可以访问代理(具有 ADDR:9092 的代理)和 kafka 数据 Node 。

当客户端连接到代理时，它说“我需要主题 XXX”。经纪人回答“主题 XXX 在 Node ADDR-YYY 上”客户端尝试连接 ADDR-YYY

可能的问题是代理返回 kafka Node 的地址，您的消费者机器无法访问该地址。它可以被阻止(例如，您没有公开特定端口)，也可以从名称无法访问(例如，损坏返回本地特定地址，如 kafka-node-1.localhost，什么时候应该是一些原始 IP 或其他东西)。