使用kafka-node连接Kafka时生产者/消费者异常问题求助

我之前也碰到过kafka-node的这类连接问题，结合你描述的场景——不管是远程Windows还是同CentOS机器都出现异常，而Kafka自带控制台工具正常，大概率是配置或库的兼容性问题，以下是一步步的排查和解决方法：

一、先确认Kafka核心配置（最常见的根源） #

Kafka默认的监听配置可能只绑定localhost，导致外部客户端（包括同机器用IP访问的情况）无法正确建立连接。你需要修改Kafka的server.properties文件：

1. 找到listeners配置，修改为：

   listeners=PLAINTEXT://0.0.0.0:9092
   

   这让Kafka监听所有网卡的9092端口。
2. 配置advertised.listeners为你的服务器公网/内网IP：

   advertised.listeners=PLAINTEXT://10.0.0.55:9092
   

   这个参数是Kafka告诉客户端用来连接的地址，如果不配置，客户端可能拿到localhost，导致连接超时。
3. 重启Kafka服务：

   # 停止Kafka
   ./kafka-server-stop.sh
   # 启动Kafka
   ./kafka-server-start.sh -daemon ../config/server.properties
   

二、调整kafka-node的客户端超时与重试参数 #

kafka-node默认的30秒请求超时可能不足以应对网络延迟或Kafka初始化的情况，你可以在创建KafkaClient时增加这些参数：

const kclient = new kafka.KafkaClient({
  kafkaHost: '10.0.0.55:9092',
  requestTimeout: 60000, // 延长请求超时到60秒
  connectTimeout: 10000, // 连接超时时间
  retryDelay: 2000, // 重试间隔
  retries: 5 // 最大重试次数
});

三、优化消费者的初始化与配置 #

你的消费者代码在client ready事件里创建，但可能缺少必要的配置项，建议调整消费者的初始化代码，增加关键配置：

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

const kclient = new kafka.KafkaClient({
  kafkaHost: '10.0.0.55:9092',
  requestTimeout: 60000,
  connectTimeout: 10000
});

kclient.on('ready',() => {
 console.log(`kclient ready`);
 // 增加消费者配置
 kconsumer = new consumer(kclient,
   [{ topic:'test', partition:0 }],
   {
     autoCommit: true,
     fetchMaxWaitMs: 1000, // 最长等待1秒返回消息
     fetchMinBytes: 1, // 有消息就返回
     sessionTimeout: 30000 // 会话超时
   }
 );
 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: ${JSON.stringify(msg)}`); })
 })
})
kclient.on('error',(err) => { console.error(`err in kclient: \n${err}\n`) })

另外，先确认你的test主题确实存在分区0：

./kafka-topics.sh --describe --topic test --zookeeper localhost:2181

如果分区0不存在，要么创建主题时指定分区数，要么修改消费者订阅的分区。

四、检查kafka-node版本兼容性 #

旧版本的kafka-node可能和新版Kafka（比如2.0以上版本）存在兼容性问题，建议升级到最新稳定版：

npm uninstall kafka-node
npm install kafka-node@latest

五、排查防火墙与网络问题 #

即使是同CentOS机器，也要确认9092端口是否开放：

1. 临时关闭防火墙测试：

   systemctl stop firewalld
   

2. 如果测试正常，重新开启防火墙并添加端口规则：

   systemctl start firewalld
   firewall-cmd --add-port=9092/tcp --permanent
   firewall-cmd --reload
   

验证步骤 #

做完以上调整后，按以下顺序测试：

1. 用Kafka自带的控制台生产者发送消息：

   ./kafka-console-producer.sh --broker-list 10.0.0.55:9092 --topic test
   

2. 运行你的kafka-node消费者代码，看是否能收到消息；再运行生产者代码，验证消息发送是否稳定。

内容的提问来源于stack exchange，提问作者yishain11