MongoDB 的 Change Streams 允许应用程序实时订阅数据库的变化。它们可以监听集合、数据库或整个部署中的插入、更新、删除和替换操作。
文章导航
一、如何创建change stream消息流
1. 创建 Change Stream
你可以在集合、数据库或整个部署级别上创建 Change Stream。以下是监听集合变化的示例：
const collection = client.db('mydatabase').collection('mycollection');
const changeStream = collection.watch();
changeStream.on('change', (change) => {
console.log('Change detected:', change);
});2. 处理 Change Events
Change Streams 会返回各种类型的事件，如 insert, update, replace, delete, 和 invalidate。你可以根据需要处理这些事件。例如：
changeStream.on('change', (change) => {
switch (change.operationType) {
case 'insert':
console.log('Document inserted:', change.fullDocument);
break;
case 'update':
console.log('Document updated:', change.updateDescription);
break;
case 'replace':
console.log('Document replaced:', change.fullDocument);
break;
case 'delete':
console.log('Document deleted:', change.documentKey);
break;
default:
console.log('Other change:', change);
}
});3. 关闭 Change Stream 和数据库连接
当你不再需要监听变化时，可以关闭 Change Stream 和数据库连接：
async function close() {
await changeStream.close();
await client.close();
}
process.on('SIGINT', close);
process.on('SIGTERM', close);二、有什么局限性
MongoDB Change Streams 是一个强大的工具，但它也有一些局限性和需要注意的事项。以下是一些主要的局限性：
1. MongoDB 版本要求
Change Streams 需要 MongoDB 3.6 或更高版本，并且只能在副本集或分片集群上使用。在单节点 MongoDB 实例上不可用。
2. 保留时间限制
Change Streams 依赖于 MongoDB 的 oplog（操作日志）。oplog 有一个有限的保留时间，通常是几天到几周，具体取决于 oplog 的大小和数据库的写入负载。如果应用程序停机或长时间未读取 Change Stream，可能会错过一些更改。
3. 操作类型限制
Change Streams 仅支持以下操作类型：insert, update, replace, delete, 和 invalidate。它们不会捕捉某些操作，比如索引创建/删除、集合重命名等。
4. 性能影响
Change Streams 会消耗额外的资源，包括 CPU、内存和网络带宽，尤其是在高负载或高吞吐量的系统中。这可能会对数据库的性能产生一定影响。
5. 安全性和权限
使用 Change Streams 需要相应的权限。用户需要有对目标数据库或集合的读取权限。另外，如果 Change Streams 被滥用，可能会导致敏感数据泄露。
6. 过滤和投影限制
虽然 Change Streams 支持使用 $match 和 $project 操作符进行过滤和投影，但这些操作符的功能有限。复杂的过滤和投影需求可能需要在应用程序层面额外处理。
7. 连接稳定性
与所有基于流的系统一样，Change Streams 对连接的稳定性有较高要求。如果网络连接不稳定，可能会导致 Change Stream 中断，应用程序需要具备重连和恢复机制。
8. 数据丢失风险
如果应用程序没有正确处理 Change Stream 的恢复（例如，存储和恢复 resume token），在 MongoDB 重启、oplog 回滚等情况下可能会丢失数据变化通知。
9. 副本集和分片集群限制
在分片集群环境中，Change Streams 可能会变得复杂，因为它们需要跨所有分片监视变化。确保所有分片都正确配置和同步非常重要。
10. 不支持事务
Change Streams 只能捕捉独立的数据库操作，不支持跨多个文档或集合的事务。如果你需要监控事务中的变化，可能需要其他机制来补充。
11. 延迟问题
虽然 Change Streams 是接近实时的，但在高负载情况下可能会有一定的延迟。如果应用程序对实时性有严格要求，需要进行性能测试和优化。
三、在什么场景下应用
1. 实时数据同步
多数据库同步
Change Streams 可以用于在多个 MongoDB 实例或不同类型的数据库之间同步数据。例如，当一个主数据库发生变化时，可以使用 Change Streams 将这些变化实时同步到一个备份数据库或其他类型的数据库（如 Elasticsearch）。
缓存更新
当数据库中的数据更新时，使用 Change Streams 立即更新缓存系统（如 Redis），确保缓存中的数据与数据库保持一致，从而提高数据访问速度。
2. 事件驱动架构
Change Streams 可以用于构建事件驱动的系统。在数据库发生变化时触发特定的业务逻辑或工作流。例如：
订单处理
在电商平台中，当一个新的订单被插入数据库时，可以使用 Change Streams 触发订单处理流程，包括库存检查、支付处理和订单确认。
用户活动追踪
在社交媒体或应用中，可以使用 Change Streams 追踪用户活动（如发布、点赞、评论），并触发相应的分析和推荐系统。
3. 实时分析和监控
实时仪表盘
使用 Change Streams 将数据变化实时推送到前端仪表盘，提供实时数据分析和监控能力。例如，实时监控网站流量、销售数据或其他关键业务指标。
异常检测
通过监控数据库中的异常数据变化（如异常登录、异常交易），及时发现和响应潜在的安全威胁或系统故障。
4. 通知系统
实时通知
在社交网络、即时通讯或其他需要实时通知的系统中，使用 Change Streams 在数据变化时立即推送通知给用户。例如，新消息通知、好友请求通知等。
工作流自动化
在企业应用中，使用 Change Streams 监控特定数据变化并触发自动化工作流，如审批流程、任务分配等。
5. 数据审核和日志记录
审计日志
使用 Change Streams 记录数据库中的所有变化，生成审计日志，用于合规性检查和故障排查。每当数据库记录发生变化时，自动生成对应的日志条目。
数据恢复
通过 Change Streams 记录的数据变化，可以在数据丢失或损坏时进行数据恢复，确保业务连续性。
6. 数据集成
微服务通信
在微服务架构中，使用 Change Streams 实现不同服务之间的数据同步和通信。例如，当用户服务中的用户数据发生变化时，通知其他相关服务（如订单服务、推荐服务）进行相应更新。
ETL（Extract, Transform, Load）
使用 Change Streams 实时提取和转换数据，将其加载到数据仓库或数据湖中，用于大数据分析和机器学习模型训练。
四、总结
MongoDB Change Streams 是一个非常灵活和强大的工具，适用于各种需要实时数据处理和响应的应用场景。通过了解和利用 Change Streams 的功能，开发者可以构建高效、响应迅速和可靠的应用程序。但是也有一定的局限性，尽管有这些局限性，MongoDB 的 Change Streams 仍然是一个非常有用的工具，特别是在需要实时监控和响应数据库变化的应用场景中。了解和应对这些局限性可以更有效地利用 Change Streams。
延展阅读：
MongoDB 4.0至7.0：这些主版本更新带来了哪些关键特性与性能飞跃？
Mongo性能优化实战：如何通过WiredTiger引擎提升MongoDB的性能和数据安全性
中小团队怎么基于PG快速迭代创新？PostgreSQL is all you need！
Nodejs 的事件循环机制是如何处理 timers、poll 和 check 队列的？
咨询方案
获取更多方案详情
