MongoDB Change Stream之一——上手及初体验

// 指定为监听某个db的修改，最多阻塞等待1分钟；
// 如果是WT4.0版本，可选择监听整个集群或者coll维度，分别为db.getMongo().watch()和db.coll.watch()
>db.watch([],{maxAwaitTimeMS:60000})

>db.test.insert({a:2})
>db.test.update({a:2}, {$set:{"msg":"hello world"}})
>db.test.insert({a:1})
>db.test.remove({a:1})
// DDL：rename/drop/dropDatabase
>db.test.renameCollection("test1")
>db.test1.drop()
>db.dropDatabase()

{ "_id" : { "_data" : "825F156B3F0000000229295A1004C982483732384D28AE57C6500C6018BF46645F696400645F156B3F0DE1FAAEF1B3DF830004" }, "operationType" : "insert", "clusterTime" : Timestamp(1595239231, 2), "fullDocument" : { "_id" : ObjectId("5f156b3f0de1faaef1b3df83"), "a" : 2 }, "ns" : { "db" : "phoenix", "coll" : "test" }, "documentKey" : { "_id" : ObjectId("5f156b3f0de1faaef1b3df83") } }
{ "_id" : { "_data" : "825F156B5F0000000129295A1004C982483732384D28AE57C6500C6018BF46645F696400645F156B3F0DE1FAAEF1B3DF830004" }, "operationType" : "update", "clusterTime" : Timestamp(1595239263, 1), "ns" : { "db" : "phoenix", "coll" : "test" }, "documentKey" : { "_id" : ObjectId("5f156b3f0de1faaef1b3df83") }, "updateDescription" : { "updatedFields" : { "msg" : "hello world" }, "removedFields" : [ ] } }
{ "_id" : { "_data" : "825F156B640000000129295A1004C982483732384D28AE57C6500C6018BF46645F696400645F156B640DE1FAAEF1B3DF840004" }, "operationType" : "insert", "clusterTime" : Timestamp(1595239268, 1), "fullDocument" : { "_id" : ObjectId("5f156b640de1faaef1b3df84"), "a" : 1 }, "ns" : { "db" : "phoenix", "coll" : "test" }, "documentKey" : { "_id" : ObjectId("5f156b640de1faaef1b3df84") } }
{ "_id" : { "_data" : "825F156B6B0000000129295A1004C982483732384D28AE57C6500C6018BF46645F696400645F156B640DE1FAAEF1B3DF840004" }, "operationType" : "delete", "clusterTime" : Timestamp(1595239275, 1), "ns" : { "db" : "phoenix", "coll" : "test" }, "documentKey" : { "_id" : ObjectId("5f156b640de1faaef1b3df84") } }
{ "_id" : { "_data" : "825F156B960000000129295A1004C982483732384D28AE57C6500C6018BF04" }, "operationType" : "rename", "clusterTime" : Timestamp(1595239318, 1), "ns" : { "db" : "phoenix", "coll" : "test" }, "to" : { "db" : "phoenix", "coll" : "test1" }, }
{ "_id" : { "_data" : "825F156BC60000000129295A1004C982483732384D28AE57C6500C6018BF04" }, "operationType" : "drop", "clusterTime" : Timestamp(1595239366, 1), "ns" : { "db" : "phoenix", "coll" : "test1" } }{ "_id" : { "_data" : "825F156BD200000001292904" }, "operationType" : "dropDatabase", "clusterTime" : Timestamp(1595239378, 1), "ns" : { "db" : "phoenix" } }

//由于使用的是db维度的watch，没法watch整个db的dropDatabase操作，因此被视为invalidate事件。
{ "_id" : { "_data" : "825F156BD200000001292904" }, "operationType" : "invalidate", "clusterTime" : Timestamp(1595239378, 1) }// 如果是WT3.6版本的话，rename执行完之后游标就中止了。这是因为在WT3.6的版本中还不支持rename事件。{ "_id" : { "_data" : BinData(0,"gl8NTwAAAAABFFoQBId+ETlGUEjUgfPVNXJ4WSsE") }, "operationType" : "invalidate" }
// 如果继续执行"it"会提示如下信息，表示changeStream生成的cursor确实已经中止了。
no cursor

//我们指定第一条，就可以获得后面的两条
>db.changestream.watch([],{maxAwaitTimeMS:60000, resumeAfter:{"_data":"825F156B3F0000000229295A1004C982483732384D28AE57C6500C6018BF46645F696400645F156B3F0DE1FAAEF1B3DF830004"}})

// Start Changestream Example 4
        pipeline := mongo.Pipeline{bson.D{{"$match", bson.D{{"$or",
            bson.A{
                bson.D{{"fullDocument.username", "alice"}},
                bson.D{{"operationType", "delete"}}}}},
        }}}
        cs, err := coll.Watch(ctx, pipeline)
        require.NoError(t, err)
        defer cs.Close(ctx)
        ok := cs.Next(ctx)
        next := cs.Current
// End Changestream Example 4

{
        _id : { // 存储元信息
            "_data" : <BinData|hex string> // resumeToken，用于断点恢复
        },
        "operationType" : "<operation>", // insert, delete, replace, update, drop, rename, dropDatabase, invalidate,部分仅支持4.0后的版本，详情见下
        "fullDocument" : { <document> }, // 修改后的数据，出现在insert, replace, delete, update的事件中
        "ns" : { // namespace
            "db" : "<database>",
            "coll" : "<collection"
        },
        "to" : { // 只在operationType为rename的时候有效，表示改名以后的namespace
            "db" : "<database>",
            "coll" : "<collection"
        },
        "documentKey" : { "_id" : <value> }, // 相当于o2字段。出现在insert, replace, delete, update事件中。正常只包含_id，对于sharded collection，还包括shard key
        "updateDescription" : { // 只在operationType为update的时候出现，相当于是增量的修改，而replace是替换
            "updatedFields" : { <document> }, // 更新的field的值
            "removedFields" : [ "<field>", ... ] // 删除的field列表
        },
        "clusterTime" : <Timestamp>, // 相当于ts字段
        "txnNumber" : <NumberLong>, // 相当于oplog里面的txnNumber，只在事务里面出现。事务号在一个事务里面单调递增
        "lsid" : { // 相当于lsid字段，只在事务里面出现。logic session id，请求所在的session的id
            "id" : <UUID>,
            "uid" : <BinData>
        }
    }