MongoDB changeStream 的 PHP 实现

什么是mongodb的changeStream

• changeStream是mongodb基于oplog(操作日志)的实时的数据变更流，应用程序可以通过changeStream实现对mongo数据变更的订阅，我们可以理解为MySQL中的触发器或者基于binlog复制的从节点。

如何使用

• changeSteam可以对某个db/colletion/deployment进行监听(watch)
• 监听的事件(event)包括insert Event、Update Event、Replace Event、 delete Event等事件

使用场景

• 可以利用这个特性实现对mongo中数据变更的监听，进行数据同步、复制等操作，举个栗子，公司原始系统使用PHP+MySQL进行服务，后续经过发展，账号服务越来越复杂，账号属性多变，数据存储切换至MongoDB，采用Java进行新服务提供，但是老系统又不能一时砍掉，砍掉我们就要丢饭碗了，那数据流从mongo到MySQL如何进行同步呢，一个方式就是可以利用changeStream 进行。

代码实现

• 这里我通过一个简单的demo来学习changeStream

• 前提准备

• 安装MongoDB

• mongodb需要开启复制集，也就是必须要主从，部分版本需要开启 enableMajorityReadConcern: true

• 安装PHPmongdoDB扩展，注意事项mongodb扩展而不是mongo

• composer 安装php-mongodb包

实现 具体实现后面贴图

• 配置mongodb的复制集, mongo.conf 在代码实例中，这里我们使用一主两从
• 启动 mongod -f /Users/gaoz/project13/mongo/db1/mongod.conf
• 启动监听客户端 php index.php
• 通过mongoshell进行数据insert/update操作

注意事项

• 在实际的使用过程中，要注意数据丢失的情况
• 考虑大量数据变更的发生，加入Mq中间件进行消息处理
• 做好服务故障监听

启动mongod服务，这里创建三个/mongo/db 分别为db1, db2, db3, 端口分别为28017,28018,28019

启动mongo监听脚本

通过mongoshell 操作数据

更新数据

监听脚本接受到数据变更

看一下接受到数据

这里重点关注三个内容

• operationType: 操作类型，可能是update、insert、replace等
• fullDocument: 文档内容
• 如果是更新的话，updateFields则是变更内容

后续的使用，比如同步到MySQL，大家可以在监听脚本中加入同步逻辑即可。

参考资料

• 官方文档:docs.mongodb.com/manual/changeStre...
• php-mongodb扩展 pecl.php.net/package/mongodb
• php-mongodb composer包 github.com/mongodb/mongo-php-libra...

参考代码

个人学习使用，欢迎指正讨论。

本作品采用《CC 协议》，转载必须注明作者和本文链接