1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

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

    }