PyMongo 变更流恢复令牌

示例说明

变更流消费者最怕两件事：重启后从头重放海量事件，或静默漏掉停机窗口内的事件。resume token 是变更流的位点凭证——每个事件都携带，把它持久化后重启时传入 resume_after，游标从断点续读，既不重放也不遗漏。管道在服务端过滤事件类型，网络只传关心的部分。

技术要点

• resume_after 从持久化令牌续读
• full_document 取代二次查询
• 令牌随每个事件即时落盘

代码解读

load 与 save 两个小函数把令牌管理收敛到文件读写；listen 的 watch 传入管道与恢复令牌，循环里每个事件先提取关键字段再保存最新令牌——先处理再存令牌的顺序保证崩溃时最多重放一个事件。插入一条订单演示事件捕获。

使用建议

令牌文件要与状态存储同级别可靠，单机文件仅演示。处理逻辑必须幂等容忍至多重放一次。变更流需要副本集环境。

代码示例

import json

from pymongo import MongoClient
from pymongo.collection import Collection

client = MongoClient('mongodb://localhost:27017', serverSelectionTimeoutMS=3000)
orders: Collection = client['shop']['orders']


def load_token() -> dict | None:
    try:
        with open('.resume-token.json') as fh:
            return json.load(fh)
    except FileNotFoundError:
        return None


def save_token(token: dict) -> None:
    with open('.resume-token.json', 'w') as fh:
        json.dump(token, fh)


def listen(max_events: int = 5) -> list[dict]:
    events: list[dict] = []
    pipeline = [{'$match': {'operationType': {'$in': ['insert', 'replace']}}}]

    with orders.watch(
        pipeline, full_document='updateLookup', resume_after=load_token()
    ) as stream:
        for event in stream:
            doc = event.get('fullDocument', {})
            events.append({
                'op': event['operationType'],
                'id': str(doc.get('_id', '')),
                'status': doc.get('status', ''),
            })
            save_token(stream.resume_token)
            if len(events) >= max_events:
                break
    return events


orders.insert_one({'status': 'created', 'total': 99})
print('captured events:', listen(max_events=1))