一、为什么需要MongoDB和Kafka集成
现代应用中，数据往往需要实时流动。比如电商平台的订单数据，既要存入数据库持久化，又要实时推送给风控系统进行分析。这时候，MongoDB作为灵活的文档数据库，和Kafka作为高吞吐的消息队列，就成了黄金搭档。
MongoDB适合存储非结构化数据，而Kafka擅长处理数据流。把它们结合起来，就能搭建一个"数据入库即流动"的管道。想象一下：用户下单后，订单数据先进入MongoDB，同时Kafka立即把这个变动通知给库存系统、物流系统，整个过程在毫秒级完成。
二、基础集成方案
我们用Node.js技术栈演示最基础的集成方式：通过监听MongoDB的变更流(Change Stream)，把变化事件推送到Kafka。
// 技术栈：Node.js + mongodb + kafkajs
const { MongoClient } = require('mongodb');
const { Kafka } = require('kafkajs');
// 1. 连接MongoDB
const mongoClient = new MongoClient('mongodb://localhost:27017');
await mongoClient.connect();
const collection = mongoClient.db('shop').collection('orders');
// 2. 配置Kafka生产者
const kafka = new Kafka({
brokers: ['localhost:9092']
});
const producer = kafka.producer();
// 3. 监听变更流并发送到Kafka
const changeStream = collection.watch();
changeStream.on('change', async (change) => {
await producer.send({
topic: 'order-updates',
messages: [
{ value: JSON.stringify(change) }
]
});
console.log(`已发送变更事件: ${change.operationType}`);
});
// 启动生产者
await producer.connect();
这个示例做了三件事：
- 连接MongoDB并监听orders集合的变更
- 配置Kafka生产者
- 把每个变更事件实时推送到order-updates主题
三、进阶处理技巧
基础方案有个问题：如果MongoDB变更很频繁，直接推送原始数据会导致Kafka主题消息爆炸。我们改进一下，加入数据过滤和格式化：
// 技术栈：Node.js (续前例)
changeStream.on('change', async (change) => {
// 只处理新增和更新操作
if (['insert', 'update'].includes(change.operationType)) {
const doc = change.fullDocument ||
await collection.findOne({ _id: change.documentKey._id });
// 构造精简消息体
const message = {
eventId: change._id.toString(),
type: 'ORDER_' + change.operationType.toUpperCase(),
payload: {
orderId: doc._id,
amount: doc.totalAmount,
userId: doc.userId
}
};
await producer.send({
topic: 'order-events',
messages: [{ value: JSON.stringify(message) }]
});
}
});
改进点包括：
- 过滤删除操作，只关注新增/更新
- 获取完整文档数据
- 构造业务友好的消息格式
- 添加事件类型标识
四、消费者端处理
现在看看Kafka消费者如何消费这些事件。我们实现一个库存扣减服务：
// 技术栈：Node.js + kafkajs
const consumer = kafka.consumer({ groupId: 'inventory-service' });
await consumer.connect();
await consumer.subscribe({ topic: 'order-events' });
await consumer.run({
eachMessage: async ({ message }) => {
const event = JSON.parse(message.value);
if (event.type === 'ORDER_INSERT') {
// 模拟库存扣减
console.log(`扣减库存 for 订单 ${event.payload.orderId}`);
await updateInventory(event.payload.items);
}
}
});
async function updateInventory(items) {
// 实际项目中这里会操作数据库
items.forEach(item => {
console.log(`商品 ${item.productId} 扣减 ${item.quantity}件`);
});
}
五、生产环境注意事项
- 错误处理：网络波动时要有重试机制
producer.on('producer.network.request_timeout', () => {
console.warn('网络超时，正在重试...');
});
- 性能优化：批量发送消息
// 每100ms或积攒100条消息时批量发送
const batchProducer = kafka.producer({
batch: {
maxSize: 100,
interval: 100
}
});
- 数据一致性：考虑使用MongoDB事务确保变更和消息发送的原子性
六、适用场景与优缺点
典型应用场景：
- 实时用户行为分析
- 跨系统数据同步
- 事件溯源架构
- 微服务间通信
优势：
- 解耦数据生产与消费
- 支持多消费者并行处理
- 借助Kafka保证消息顺序
局限：
- 需要维护两个系统
- 变更流可能丢失短暂连接期间的事件
- 需要处理重复消息(幂等消费)
七、总结
MongoDB和Kafka的集成就像给数据安上了翅膀。MongoDB作为数据的家，Kafka作为数据的传送带，两者配合让数据既能安全存储，又能实时流动。关键在于：
- 合理设计消息格式
- 处理好错误和重试
- 根据业务特点选择同步策略
这种架构特别适合需要快速响应数据变化的场景，但也需要根据业务规模权衡复杂度。对于中小项目，可以先从简单方案开始，随着业务增长逐步完善。
评论