剖析后OpLog订阅MongoDB的数据变更就没那么难了

private BsonTimestamp queryTs; @Test public void OpLogTest() { MongoClient mongoClient = new MongoClient(new MongoClientURI("mongodb://admin:admin@127.0.0.1:3717")); MongoCollectioncollection = mongoClient.getDatabase("local") .getCollection("oplog.rs"); //如果是首次订阅，需要使用自然排序查询，获取第最后一次操作的操作时间戳。如果是续订阅直接读取记录的值赋值给queryTs即可 FindIterabletsCursor = collection.find().sort(new BasicDBObject("$natural", -1)) .limit(1); Document tsDoc = tsCursor.first(); queryTs = (BsonTimestamp) tsDoc.get("ts"); while (true) try { //构建查询语句,查询大于当前查询时间戳queryTs的记录 BasicDBObject query = new BasicDBObject("ts", new BasicDBObject("$gt", queryTs)); MongoCursordocCursor = collection.find(query) .cursorType(CursorType.TailableAwait) //没有数据时阻塞休眠 .noCursorTimeout(true) //防止服务器在不活动时间（10分钟）后使空闲的游标超时。 .oplogReplay(true) //结合query条件，获取增量数据，这个参数比较难懂，见：https://docs.mongodb.com/manual/reference/command/find/index.html .maxAwaitTime(1, TimeUnit.SECONDS) //设置此操作在服务器上的最大等待执行时间 .iterator(); while (docCursor.hasNext()) { Document document = docCursor.next(); //更新查询时间戳 queryTs = (BsonTimestamp) document.get("ts"); //TODO 在这里接收到数据后通过订阅数据路由分发 String op = document.getString("op"); String database = document.getString("ns"); Document context = (Document) document.get("o"); Document where = null; if (op.equals("u")) { where = (Document) document.get("o2"); if (context != null) { context = (Document) context.get("$set"); } } System.err.println("操作时间戳：" + queryTs.getTime()); System.err.println("操作类  型：" + op); System.err.println("数据库.集合：" + database); System.err.println("更新条件：" + JSON.toJSONString(where)); System.err.println("文档内容：" + JSON.toJSONString(context)); } } catch (Exception e) { e.printStackTrace(); } }