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");

        //如果是首次訂閱，需要使用自然排序查詢，獲取第最后一次操作的操作時間戳。如果是續(xù)訂閱直接讀取記錄的值賦值給queryTs即可
        FindIterabletsCursor = collection.find().sort(new BasicDBObject("$natural", -1))
                .limit(1);
        Document tsDoc = tsCursor.first();
        queryTs = (BsonTimestamp) tsDoc.get("ts");
        while (true) try {
            //構(gòu)建查詢語句,查詢大于當(dāng)前查詢時間戳queryTs的記錄
            BasicDBObject query = new BasicDBObject("ts", new BasicDBObject("$gt", queryTs));
            MongoCursordocCursor = collection.find(query)
                    .cursorType(CursorType.TailableAwait) //沒有數(shù)據(jù)時阻塞休眠
                    .noCursorTimeout(true) //防止服務(wù)器在不活動時間（10分鐘）后使空閑的游標(biāo)超時。
                    .oplogReplay(true) //結(jié)合query條件，獲取增量數(shù)據(jù)，這個參數(shù)比較難懂，見：https://docs.mongodb.com/manual/reference/command/find/index.html
                    .maxAwaitTime(1, TimeUnit.SECONDS) //設(shè)置此操作在服務(wù)器上的最大等待執(zhí)行時間
                    .iterator();
            while (docCursor.hasNext()) {
                Document document = docCursor.next();
                //更新查詢時間戳
                queryTs = (BsonTimestamp) document.get("ts");
                //TODO 在這里接收到數(shù)據(jù)后通過訂閱數(shù)據(jù)路由分發(fā)

                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("數(shù)據(jù)庫.集合：" + database);
                System.err.println("更新條件：" + JSON.toJSONString(where));
                System.err.println("文檔內(nèi)容：" + JSON.toJSONString(context));
            }
        } catch (Exception e) { e.printStackTrace(); }
    }