KafkaJS 主题生产与消费

示例说明

Kafka 的吞吐与可靠性来自分区与位移两个概念。消息带 key 时同 key 落入同一分区，单分区内严格有序——同一客户的订单事件不会乱序消费。手动提交位移把确认权交还业务代码：只有处理成功才提交，崩溃重启后从上次提交位置继续，配合幂等生产者实现至少一次投递下的不丢不乱。

技术要点

• 消息 key 决定分区与局部有序
• 手动提交位移绑定处理成功语义
• 心跳维持消费者组的会话活跃

代码解读

生产者开启幂等模式防重试重排序，两条同 key 消息演示分区亲和；消费者的 eachMessage 里解析订单并逐条提交位移加一，heartbeat 防止长处理被组协调器判定失联。SIGTERM 处理器按消费者先、生产者后的顺序断开，排空在途消息。

使用建议

处理逻辑必须幂等，重复投递是常态而非异常。分区数要预留扩展余量。lag 监控是消费延迟的哨兵指标，纳入告警。

代码示例

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'orders-service',
  brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
  logLevel: logLevel.WARN,
});

const producer = kafka.producer({ idempotent: true });
await producer.connect();

await producer.send({
  topic: 'orders.created',
  messages: [
    { key: 'customer-42', value: JSON.stringify({ orderId: 'ord-1', total: 99 }) },
    { key: 'customer-42', value: JSON.stringify({ orderId: 'ord-2', total: 12 }) },
  ],
});

const consumer = kafka.consumer({ groupId: 'billing-group' });
await consumer.connect();
await consumer.subscribe({ topic: 'orders.created', fromBeginning: false });

await consumer.run({
  autoCommit: false,
  eachBatchAutoResolve: false,
  eachMessage: async ({ topic, partition, message, heartbeat }) => {
    const order = JSON.parse(message.value?.toString() ?? '{}');

    console.log('billing for', order.orderId,
      'partition', partition, 'offset', message.offset);

    await heartbeat();
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

process.on('SIGTERM', async () => {
  await consumer.disconnect();
  await producer.disconnect();
});