Oplog 是 MongoDB 實現(xiàn)復制集的關鍵數(shù)據(jù)結構，在復制集中 Primary 對數(shù)據(jù)庫操作之后就會產(chǎn)生一個 Oplog 文檔保存在 local.oplog.rs 集合中，Secondary 成員會拉取 Primary 的 Oplog 并重放相同的操作，從而達到 Secondary 成員與 Primary 有一致的數(shù)據(jù)。實際上復制集中每一個成員都會保存 Oplog，其他成員會根據(jù)連接延遲等因數(shù)選擇最近的成員拉取 Oplog 數(shù)據(jù)。

Oplog 存在集合 local.oplog.rs，這是系統(tǒng)內(nèi)置集合，一個 capped collection，即是這個 collection 有固定大小，一旦寫滿數(shù)據(jù)會從頭開始寫入，就像一個圓形的隊列結構。這個 collection 大小在初始化集群時設置，默認的大小是 5% 的空閑磁盤空間，也可以在配置文件設置 oplogSizeMB 選項，或者在啟動 MongoDB 后使用 replSetResizeOplog 命令動態(tài)設置 collection 大小。

Oplog 與 MongoDB 的其他的文檔沒有什么不同，它固定有一些屬性：

1. ts: MongoDB 的內(nèi)置的特殊時間戳數(shù)據(jù)結構，如 Timestamp(1503110518, 1), 由秒級的 Unix 時間戳和一個順序增長的整數(shù) increment 表示。長度為 64 位，其中 Unix 時間戳占 32 位，后 32 位可以保存同一秒內(nèi)的第幾次操作。
2. h: hash 值代表每個 Oplog 的唯一標識。
3. v: Oplog 版本
4. ns: namespace 命名空間，數(shù)據(jù)庫+集合，用 database.collection 表示。但如果是表操作命令等，變成 database.$cmd。
5. op：operation type，操作類型，包含以下幾種：
   1. i: insert, 插入文檔
   2. u: update, 更新文檔
   3. d: delete, 刪除文檔
   4. c: command, 操作命令，如 createIndex 等
   5. n: 空操作，用于空閑時主從同步 Oplog 時間信息
6. o: operation， Oplog 操作的具體內(nèi)容，例如 i operation type，o 即是插入的文檔。對于 u operation type, 只更新部分內(nèi)容， o 鍵的內(nèi)容為 {$set: {...}}
7. o2: 用于 update 操作，包含 _id 屬性值。

Oplog 的重放是冪等(idempotent)的，即是說同一個 Oplog 重放多次最終結果還是一致的。這是 MongoDB 將許多命令操作進行了轉(zhuǎn)化，保持生成的 Oplog 是可以冪等的，如執(zhí)行以下 $inc 操作：

db.test.update({_id: ObjectId("533022d70d7e2c31d4490d22")}, {$inc: {count: 1}})

產(chǎn)生的 Oplog 為:

{
 "ts" : Timestamp(1503110518, 1),
 "t" : NumberLong(8),
 "h" : NumberLong(-3967772133090765679),
 "v" : NumberInt(2),
 "op" : "u",
 "ns" : "mongo.test",
 "o2" : {
  "_id" : ObjectId("533022d70d7e2c31d4490d22")
 },
 "o" : {
  "$set" : {
   "count" : 2.0
  }
 }
}

以上 MongoDB 可以保證 Oplog 的數(shù)據(jù)操作(DML 語句)是冪等的，但數(shù)據(jù)表操作(DDL 語句)命令無法保證，例如重復執(zhí)行相同的 createIndex 命令。

Oplog 的查詢

Capped collection 內(nèi)文檔是以插入順序排序的，沒有其他索引，但是 local.oplog.rs 是一個特殊的 capped collection，在 Wiredtiger 引擎的話，Oplog 的時間戳會作為一個特殊的元信息存儲，使得 Oplog 可以以 ts 字段排序，查詢 Oplog 時可以利用 ts 字段篩選。

一般來說 Secondary 同步需要經(jīng)過 initial sync 和 incremental sync，initial sync 同步完成后，需拉取從同步時間點開始之后的 Oplog 進行持續(xù)重放。所以查詢 Oplog 的操作一般是：

db.oplog.rs.find({$gte:{'ts': Timestamp(1503110518, 1)}})

Secondary 需要不斷獲取 Primary 產(chǎn)生的 Oplog, 復制集會使用 tailable cursor 持續(xù)獲取 Oplog 數(shù)據(jù)，非常類似 Unix 系統(tǒng)的 tail -f。這會提高效率，因為一般的 cursor 使用完畢后就會關閉，而 tailable cursor 會保存上次的 id, 并持續(xù)獲取數(shù)據(jù)。

如果使用 pymongo 驅(qū)動器，則定位從某個時間點之后的 Oplog 可以這麼寫：

coll = db['local'].get_collection(
 'oplog.rs',
 codec_options=bson.codec_options.CodecOptions(document_class=bson.son.SON))

cursor = coll.find({'ts': {'$gte': start_optime}},
 cursor_type=pymongo.cursor.CursorType.TAILABLE,
 oplog_replay=True,
 no_cursor_timeout=True)

while True:
 try:
  oplog = cursor.next()
  process(oplog)
 except StopException:
  # 沒有更多的 Oplog 數(shù)據(jù)
  time.sleep(1)

cursor_type 使用 TAILABLE 或者 TAILABLE_AWAIT，使用后一種類型時，如果沒有更多的 Oplog 數(shù)據(jù)，則這次請求會阻塞等待有 Oplog 數(shù)據(jù)或者到達等待的時間超時返回。

設置 oplog_replay 標記可以表示此次請求的類型是保存 Oplog 的 capped collection, 提供 ts 篩選參數(shù), 進行查詢優(yōu)化。

獲取到 Oplog 之后，就可以做數(shù)據(jù)同步或者分發(fā)到感興趣的消費者作特殊分析，如 MongoShake 工具。

參考了文檔：

Replica Set Oplog: https://docs.mongodb.com/manual/core/replica-set-oplog/
MongoDB oplog 漫談: http://caosiyang.github.io/2016/12/24/mongodb-oplog/
MongoDB復制集原理: http://www.dbjr.com.cn/article/166148.htm

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。