Node.js Kafka 进阶

server.properties 核心配置

配置	说明	默认值
broker.id	Broker 唯一标识	0
log.dirs	日志存储目录	/tmp/kafka-logs
zookeeper.connect	Zookeeper 地址	localhost:2181
num.partitions	默认分区数	1
log.retention.hours	消息保留时间	168 (7天)

Node.js 进阶操作

指定分区发送

await producer.send({
  topic: 'orders',
  messages: [{ key: 'user-123', value: '...', partition: 0 }]
})
// key 相同 → 同一分区（保证顺序）

批量发送

await producer.send({
  topic: 'bulk',
  messages: [
    { value: 'msg1' },
    { value: 'msg2' },
    { value: 'msg3' }
  ]
})

创建主题

const admin = kafka.admin()
await admin.connect()
await admin.createTopics({
  topics: [{ topic: 'new-topic', numPartitions: 3, replicationFactor: 1 }]
})
await admin.disconnect()

消费者提交偏移量

await consumer.run({
  autoCommit: true,                  // 自动提交
  autoCommitInterval: 5000,          // 每5秒提交一次
  eachMessage: async ({ topic, partition, message }) => {
    // 处理消息...
  }
})

手动提交

autoCommit: false
// 在 eachMessage 中手动提交
await consumer.commitOffsets([{ topic, partition, offset: message.offset }])

最佳实践

实践	说明
key 设计	相同 key → 同分区 → 保证顺序
分区数	根据并发消费者决定（≤消费者数）
保留策略	按时间或大小设置合理的清理策略
压缩	启用 snappy/gzip 减少网络传输