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

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

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

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

Oplog 的重放是冪等(idempotent)的，即是說(shuō)同一個(gè) Oplog 重放多次最終結(jié)果還是一致的。這是 MongoDB 將許多命令操作進(jìn)行了轉(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 語(yǔ)句)是冪等的，但數(shù)據(jù)表操作(DDL 語(yǔ)句)命令無(wú)法保證，例如重復(fù)執(zhí)行相同的 createIndex 命令。

Oplog 的查詢

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

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

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

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

如果使用 pymongo 驅(qū)動(dòng)器，則定位從某個(gè)時(shí)間點(diǎn)之后的 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，使用后一種類型時(shí)，如果沒有更多的 Oplog 數(shù)據(jù)，則這次請(qǐng)求會(huì)阻塞等待有 Oplog 數(shù)據(jù)或者到達(dá)等待的時(shí)間超時(shí)返回。

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

獲取到 Oplog 之后，就可以做數(shù)據(jù)同步或者分發(fā)到感興趣的消費(fèi)者作特殊分析，如 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復(fù)制集原理: http://m.fzitv.net/article/166148.htm

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論