剖析後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(); }
    }