change stream:resume token 与 oplog 的关系

事件与 token

实测 · 本机执行输出

  insert  ns={'db': 'test', 'coll': 't'}  documentKey={'_id': 1}  clusterTime=Timestamp(1791775093, 20)  _id._data=826ACC5175000000142B042C0100…
  insert  ns={'db': 'test', 'coll': 't'}  documentKey={'_id': 2}  clusterTime=Timestamp(1791775093, 20)  _id._data=826ACC5175000000142B042C0100…
  update  ns={'db': 'test', 'coll': 't'}  documentKey={'_id': 1}  clusterTime=Timestamp(1791775093, 21)  _id._data=826ACC5175000000152B042C0100…
  事件的键: ['_id', 'clusterTime', 'documentKey', 'fullDocument', 'ns', 'operationType', 'wallTime']

每个事件的 _id 就是 resume token。同一次批量写入的多个事件 clusterTime 相同,但 token 不同。

token 编码了 oplog 位置

实测 · 本机执行输出

  token 826ACC5175000000142B042C0100296E5A100411…  首字节 0x82;后 8 字节解码为 Timestamp(1791775093, 20);事件的 clusterTime = Timestamp(1791775093, 20);相等: True
  token 826ACC5175000000142B042C01002B026E5A1004…  首字节 0x82;后 8 字节解码为 Timestamp(1791775093, 20);事件的 clusterTime = Timestamp(1791775093, 20);相等: True
  token 826ACC5175000000152B042C0100296E5A100411…  首字节 0x82;后 8 字节解码为 Timestamp(1791775093, 21);事件的 clusterTime = Timestamp(1791775093, 21);相等: True
  token 总长 83 字节

_data 的首字节 0x82 后面 8 个字节就是 clusterTime(秒加序号),实测与事件的 clusterTime 完全一致。

恢复

实测 · 本机执行输出

  从第 2 个事件的 token 继续:收到 1 个事件: update 1(正是原来的第 3 个事件)
  之后新写入也能收到: ['insert 3']

用 resumeAfter 传第 N 个事件的 token,从第 N+1 个事件继续,不丢不重;startAtOperationTime 则从一个时间点开始(含该时间点的写入)。

空闲 stream

实测 · 本机执行输出

  从 operationTime=Timestamp(1791775096, 1)(它是当时最后一次写入的时间)开始,收到: [3, 100, 101](含该时间点的那次写入 _id=3,以及之后的写入)

没有匹配事件时,stream 的 resume_token 仍随 oplog 前进(postBatchResumeToken),所以用它恢复不会因 token 太旧而失败。

token 太旧

实测 · 本机执行输出

  token 前: 826ACC5179000000032B0429296E14…;其它集合写了 1000 条后(本 stream 没有事件): 826ACC5179000000052B0429296E14…;token 前进了: True

token 对应的位置不在节点的 oplog 里就无法恢复:新加入的成员 oplog 只从同步点开始,用旧 token 在它上面恢复报 286。

怎么避免

实测 · 本机执行输出

  新成员状态 SECONDARY(initial sync 用了 2 s)
  primary 的 oplog 最早一条 ts = Timestamp(1791775093, 1);新成员的 oplog 最早一条 ts = Timestamp(1791775102, 6)(比 primary 晚);最旧事件 token 的 clusterTime = Timestamp(1791775093, 20)
  在新成员(secondary 读偏好)上用旧 token 恢复:code=286(ChangeStreamHistoryLost);Executor error during aggregate command on namespace: test.t :: caused by :: Resume of change stream was not possible, as the resume point may no long
  在 primary 上用同一个旧 token 恢复:成功,收到的第一个事件 insert {'_id': 2}

要及时保存最新的 token,并保证 oplog 窗口(见后面 oplog 窗口一篇)足够覆盖消费者可能停机的时间。

分步演示

change stream 的 resume token

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / MongoDB 9.0.2(Homebrew 的 mongod、mongos)/ mongosh 2.12.0 / pymongo 4.18.3;每个实验在临时目录里起全新的 mongod(独立 dbpath,端口 9900 以上,只监听 127.0.0.1,WiredTiger 缓存 0.25 GB),需要副本集或分片集群时在本机起多个进程;命令输出来自真实的 mongosh,批量写入、计时与多线程用 pymongo;计时用 time.perf_counter,数字每次运行会有小幅抖动,ObjectId、时间戳等每次运行都不同。单节点副本集(oplog 128 MB),第 6 节再加一个 priority 0 的成员
没有验证:primary 自己的 oplog 被覆盖后的恢复失败(我试图把 128 MB 的单节点 oplog 写到约 976 MB,最早的条目始终没有被截断,原因没有查明,所以改用新成员构造);分片集群上的 change stream token(包含各分片的位置);startAfter 与 invalidate 事件;预镜像(changeStreamPreAndPostImages)的 token

下一篇 · MONGODB 59

change stream 的过滤 pipeline 与 fullDocument