剖析后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测试数据/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(); }

     }