推荐一个强大的Node.js Kafka流处理库——node-kafka-streams

const { KafkaStreams } = require('node-kafka-streams');
 
// 配置Kafka客户端
const streamsClientConfig = {
  kafka: {
    kafkaHost: 'kafka-host:9092', // Kafka 代理地址
  },
  // 设置应用程序的全局配置
  app: {
    logLevel: 'debug', // 日志级别
    devMode: true, // 开发模式
  },
};
 
// 创建KafkaStreams实例
const streams = new KafkaStreams(streamsClientConfig);
 
// 使用高级API创建一个消费者实例
const consumer = streams.getKafkaConsumer({
  topic: 'my-topic', // 订阅的主题
});
 
// 使用高级API创建一个生产者实例
const producer = streams.getKafkaProducer();
 
// 订阅消息事件
consumer.on('message', (message) => {
  console.log(`接收到消息: ${message.value.toString()}`);
  // 处理消息逻辑...
 
  // 发送消息到另一个主题
  producer.produce({
    topic: 'my-output-topic',
    messages: `处理后的消息: ${message.value.toString()}`,
  });
});
 
// 启动消费者
consumer.start();

这段代码展示了如何使用node-kafka-streams库来创建Kafka的消费者和生产者，订阅主题并处理接收到的消息。同时，它还展示了如何发送处理后的消息到另一个主题。这个例子简洁地展示了如何在Node.js环境中使用Kafka进行消息流的处理。