1 如何实时同步MongoDB？
MongoDB 从 3.6 开始为开发者提供了 Change Streams 功能，利用 Change Streams 功能可以非常方便地监听指定 Collection 上的数据变化。例如在 mongo shell 中，我们可以通过如下方式监听 shopping 数据库 order 表上的变化：
watchCursor = db.getSiblingDB("shopping").order.watch() while (!watchCursor.isExhausted()){ if (watchCursor.hasNext()){ printjson(watchCursor.next()); } }
2 在Play中如何操作？
利用 Play Mongo 可以方便地实现数据监听功能，并且我们可以将 Change Stream 转换成 Akka Stream，然后以流的方式处理指定 Collection 上的数据变化，
mongo .collection[Order] .watch() .fullDocument .toSource .groupedWithin(10, 1000.millis) .throttle(elements = 1, per = 1.second, maximumBurst = 1, ThrottleMode.shaping) .runForeach{ seq => // ... }
上面的代码实现了以下几个功能：
- 将从 Change Stream 接收到的元素进行缓冲，以方便批处理，当满足下面任意一个条件时便结束缓冲向后传递：
- 缓冲满10个元素
- 缓冲时间超过了1000毫秒
- 对缓冲后的元素进行流控，每秒只允许通过1个元素
3 如何实现高可用？
上面的代码并没有考虑可用性，如果在监听过程中发生了网络错误，如何从错误中恢复呢？ 上面的实现代码底层是基于官方的 mongo-java-driver 实现的，关于可用性官方文档有如下描述：
Change streams provide a way to watch changes to documents in a collection. To improve the usability of this new stage, the MongoCollection API includes a new watch method. The ChangeStreamIterable sets up the change stream and automatically attempts to resume if it encounters a potentially recoverable error.
文档中提及程序可以自动从可恢复的错误中恢复。经测试验证，如果网络中断在 30 秒以内均属于可恢复错误；但是如果大于 30 秒，则会报连接超时错误并且无法从错误中自动恢复：
com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting for a server that matches ReadPreferenceServerSelector{readPreference=primary}. Client view of cluster state is {type=UNKNOWN, servers=[{address=127.0.0.1:27117, type=UNKNOWN, state=CONNECTING, exception={com.mongodb.MongoSocketOpenException: Exception opening socket}, caused by {java.net.ConnectException: Connection refused}}] at com.mongodb.internal.connection.BaseCluster.createTimeoutException(BaseCluster.java:401) at com.mongodb.internal.connection.BaseCluster.handleServerSelectionRequest(BaseCluster.java:309) at com.mongodb.internal.connection.BaseCluster.access$800(BaseCluster.java:65) at com.mongodb.internal.connection.BaseCluster$WaitQueueHandler.run(BaseCluster.java:482) at java.lang.Thread.run(Thread.java:748)
幸运的是，Akka Stream 的 RestartSource 可以帮我们解决这种不可恢复错误，解决方式就是通过指数规避(exponential back-off)方式不断重试。下面是一个通用的创建 RestartSource 的方法实现：
def restartSource(colName: String): Source[ChangeStreamDocument[JsObject], _] = { RestartSource.withBackoff( minBackoff = 3.seconds, maxBackoff = 10.seconds, randomFactor = 0.2, maxRestarts = 1000000 ) { () ⇒ Logger.warn(s"Creating source for watching ${colName}.") mongo.collection(colName).watch().fullDocument.toSource } }
通过 Backoff 参数可以指定重试策略：
- minBackoff 最小重试时间间隔
- maxBackoff 最大重试时间间隔
- randomFactor 设置一个随机的浮动因子，使得每次计算的间隔有些许差异
- maxRestarts 最大重试次数
当发生错误时，RestartSource 会尝试重新创建一个 Source：
Logger.warn(s"Creating source for watching ${colName}.") mongo.collection(colName).watch().fullDocument.toSource
完整代码如下：
val colName = "common-user"
restartSource(colName)
.groupedWithin(10, 1000.millis)
.throttle(elements = 1, per = 1.second, maximumBurst = 1, ThrottleMode.shaping)
.runForeach{ seq =>
try {
Logger.info(seq.toString())
} catch { case t: Throwable =>
Logger.error(s"Watch change stream of ${colName} error: ${t.getMessage}", t)
}
}
需要注意的是 runForeach 中需要显式捕获异常并处理，否则会导致 Source 结束并退出。
