中文
devexamples

KafkaJS Producer and Consumer Topics

Publish and consume Kafka topics with KafkaJS: keyed partitioning for order, consumer groups with manual commit, and graceful shutdown.

TypescriptNodeData Fetching

Code

snippet
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();
});

Related Examples