PlayScala 开发技巧 - 实时同步 MongoDB 高可用方案

原创
2018/09/03 08:21
阅读数 696

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 结束并退出。

展开阅读全文
打赏
0
5 收藏
分享
加载中
更多评论
打赏
0 评论
5 收藏
0
分享
返回顶部
顶部