Windows10下NodeJS连接Kafka出现ECONNREFUSED错误求助

一、解决connect ECONNREFUSED ::1:9092错误 #

出现该错误的核心原因是Windows系统中localhost默认优先解析为IPv6地址::1，但你的Kafka Broker仅监听了IPv4的127.0.0.1，导致客户端通过IPv6地址连接失败。可通过以下两种方式解决：

1. 修改Kafka Broker配置 #

打开server.properties文件，调整以下配置：

# 明确绑定IPv4地址
listeners=PLAINTEXT://127.0.0.1:9092
# 确保对外通告的地址与监听地址一致，客户端将使用该地址建立连接
advertised.listeners=PLAINTEXT://127.0.0.1:9092

修改后重启Kafka服务。

2. 在NodeJS代码中直接使用IPv4地址 #

无需修改Kafka配置，直接在客户端代码中将localhost替换为127.0.0.1：

const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'high-concurrency-app',
  brokers: ['127.0.0.1:9092'] // 用127.0.0.1替代localhost
})

二、高并发场景下的NodeJS Kafka客户端推荐 #

1. Kafkajs（v2.0.0+） #

作为纯JavaScript实现的客户端，Kafkajs维护活跃、API简洁，性能足以应对绝大多数高并发场景，且支持现代NodeJS特性（如async/await）。通过以下配置优化高并发性能：

生产者优化配置

const producer = kafka.producer({
  allowAutoTopicCreation: false, // 禁用自动创建主题，提前创建确保稳定性
  retry: {
    retries: 5, // 重试次数
    initialRetryTime: 100 // 初始重试间隔
  },
  batch: {
    size: 16384, // 批量发送的消息大小阈值（16KB），可根据业务调整
    linger: 5 // 等待5ms凑齐批量，减少请求次数
  },
  compression: 'gzip' // 启用GZIP压缩，降低网络传输开销
})

// 批量发送示例
await producer.send({
  topic: 'test',
  messages: Array.from({ length: 1000 }, (_, i) => ({ value: `message-${i}` }))
})

消费者优化配置

const consumer = kafka.consumer({
  groupId: 'high-concurrency-group',
  sessionTimeout: 30000, // 会话超时时间
  heartbeatInterval: 3000, // 心跳间隔
  maxBytesPerPartition: 1048576, // 每个分区单次拉取的最大字节数（1MB）
  fetchMinBytes: 1024, // 拉取的最小字节数，凑够才返回
  fetchMaxWaitMs: 500 // 最长等待时间，到点即使没凑够最小字节也返回
})

// 高并发消费示例
await consumer.subscribe({ topic: 'test', fromBeginning: false })
await consumer.run({
  eachBatchAutoResolve: true,
  async eachBatch({ batch, resolveOffset, heartbeat }) {
    // 批量处理消息，减少单条处理的开销
    for (const message of batch.messages) {
      // 处理消息逻辑
      console.log(`Received message: ${message.value.toString()}`)
      resolveOffset(message.offset)
      await heartbeat()
    }
  }
})

2. node-rdkafka #

基于高性能C库librdkafka的NodeJS绑定，性能比纯JS客户端更出色，适合极致高并发场景（如每秒处理数十万条消息）。虽然API相对复杂，但底层的成熟度能保证稳定性和吞吐量。

示例配置（生产者）：

const Kafka = require('node-rdkafka')

const producer = new Kafka.Producer({
  'metadata.broker.list': '127.0.0.1:9092',
  'compression.codec': 'gzip',
  'batch.num.messages': 1000, // 批量发送的消息数
  'queue.buffering.max.ms': 5 // 缓冲队列最大等待时间
})

producer.connect()

producer.on('ready', () => {
  // 批量发送消息
  const messages = Array.from({ length: 1000 }, (_, i) => ({
    topic: 'test',
    value: Buffer.from(`message-${i}`)
  }))
  producer.produceBatch(messages, (err) => {
    if (err) console.error('Produce error:', err)
  })
})

额外优化建议 #

• 调整Kafka Broker的server.properties：提高num.network.threads（默认3）和num.io.threads（默认8）的值，比如设置为num.network.threads=5、num.io.threads=16，提升Broker的处理能力。
• 提前创建好主题，并设置合理的分区数（分区数越高，并行消费能力越强），例如：

kafka-topics.bat --create --topic test --bootstrap-server 127.0.0.1:9092 --partitions 8 --replication-factor 1

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