目录
Apache Kafka 作为高吞吐、分布式的消息队列系统，在实时数据流处理中占据重要地位。本文将以 Node.js 为例，从基础概念到代码实战，手把手教你实现 Kafka 的生产者与消费者。
一、Kafka 核心概念速览
1. Topic 与 Partition
- Topic：消息的分类（如 `userlogs`），生产者发送到 Topic，消费者订阅 Topic。
- Partition：每个 Topic 分为多个分区，实现并行处理。分区内有序，分区间无序。
- 例如：将 `userlogs` 分为 3 个分区，可同时由 3 个消费者处理。
2. Producer 与 Consumer
- Producer：向 Kafka 发送消息的客户端。
- Consumer：订阅 Topic 并处理消息，消费者组（Consumer Group） 实现负载均衡。
3. Broker 与 Cluster
- Broker：Kafka 服务节点，负责存储和转发消息。
- Cluster：多个 Broker 组成集群，通过副本机制保障高可用。
二、环境准备
1. 安装 Kafka
参考 Kafka 官方文档 启动本地 Kafka 服务（需 Zookeeper 或 KRaft 模式）。
2. Node.js 客户端库
npm install kafkajs # 推荐：轻量、API 友好
# 或使用 node-rdkafka（高性能，但配置复杂）三、生产者实现：发送消息
// producer.js
const { Kafka } = require('kafkajs');
const kafka = new Kafka({
clientId: 'node-producer',
brokers: ['localhost:9092'], // 替换为实际 Broker 地址
});
const producer = kafka.producer();
async function sendMessage() {
await producer.connect();
await producer.send({
topic: 'user_actions',
messages: [
{
key: 'user1', // 相同 Key 的消息分配到同一分区
value: JSON.stringify({ action: 'click', page: 'home' })
},
],
});
console.log('✅ 消息发送成功');
await producer.disconnect();
}
sendMessage().catch(console.error);运行命令：`node producer.js`
四、消费者实现：处理消息
// consumer.js
const { Kafka } = require('kafkajs');
const kafka = new Kafka({
clientId: 'node-consumer',
brokers: ['localhost:9092'],
});
const consumer = kafka.consumer({ groupId: 'user-analytics-group' });
async function startConsumer() {
await consumer.connect();
await consumer.subscribe({ topic: 'user_actions', fromBeginning: true });
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
console.log(`📩 收到消息:
Topic: ${topic}
Partition: ${partition}
Key: ${message.key.toString()}
Value: ${message.value.toString()}
`);
// 手动提交 Offset（确保消息处理完成）
await consumer.commitOffsets([{ topic, partition, offset: message.offset }]);
},
});
}
startConsumer().catch(console.error);运行命令：`node consumer.js`
五、高级配置与最佳实践
1. 生产者优化
const producer = kafka.producer({
idempotent: true, // 开启幂等性，防止重复消息
transactionTimeout: 30000,
});
// 批量发送提升吞吐量
await producer.send({
topic: 'logs',
messages: batchMessages,
acks: -1 // 所有副本确认后才返回成功
});2. 消费者容错处理
consumer.on('consumer.crash', (error) => {
console.error('消费者崩溃:', error);
process.exit(1); // 重启或告警
});
// 处理消息时捕获异常，避免 Offset 提交错误数据3. 安全认证（生产环境必配）
new Kafka({
brokers: ['kafka-server:9093'],
ssl: { rejectUnauthorized: false },
sasl: {
mechanism: 'scram-sha-256',
username: process.env.KAFKAUSER,
password: process.env.KAFKAPASS
}
});六、常见问题解决
•消息重复消费：消费者处理消息后崩溃，导致 Offset 未提交。
方案：实现业务逻辑的幂等性（如数据库唯一键）。
•性能瓶颈：单个消费者处理速度慢。
方案：增加分区数，启动多个消费者实例（相同 Group ID）。
•数据丢失风险：生产者配置 `acks: 0` 时，不等待 Broker 确认。
方案：生产环境至少设置 `acks: 1`（Leader 确认）。
七、应用场景示例
•用户行为追踪：Web 端埋点数据实时发送到 Kafka，消费者计算点击率。
•日志聚合：微服务日志统一写入 Kafka，供 ELK 系统分析。
•订单状态通知：订单支付成功后，通过 Kafka 触发短信通知。
总结
通过 `kafkajs`，Node.js 可快速集成 Kafka 实现高可靠的消息处理。关键点：
1. 生产者关注消息分区策略与批量发送。
2. 消费者需处理 Offset 提交与容错。
3. 生产环境务必配置 SSL 和 SASL 认证。
进一步学习：Kafka 中文学习网
