直接用find()读取副本集change stream会失败，因change stream必须在主节点建立且需watch权限和readConcern:"majority"；连从节点会报CommandNotSupportedOnSecondary或NotMaster错误。
为什么直接用 find() 读取副本集的 change stream 会失败
因为副本集的 change stream 必须在主节点上建立，且客户端需启用 watch 权限和读取关注（readConcern: "majority"）。直接对从节点调用 collection.watch() 会报 CommandNotSupportedOnSecondary 或 NotMaster 错误。PyMongo 默认可能路由到从节点，尤其当连接字符串含多个 host 且未显式指定 readPreference 时。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 连接字符串必须包含
?replicaSet=xxx参数，否则 PyMongo 不启用副本集模式 - 显式设置
read_preference=ReadPreference.PRIMARY（这是默认值，但显式写出来可防配置覆盖） - 确保 MongoDB 用户拥有
clusterMonitor或至少changeStream权限 - 避免在
watch()前手动client.admin.command("ismaster")切换节点——PyMongo 内部已自动处理主节点发现
如何正确初始化 change stream 并处理断连重试
Change stream 不是长连接保活协议，网络抖动、主从切换或 oplog 过期都会导致 StopIteration 或 OperationFailure（如 CursorNotFound、ResumableChangeStreamError）。不能只靠 for change in stream: 循环。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 始终用
try/except捕获pymongo.errors.PyMongoError，特别关注ConnectionFailure、OperationFailure和InvalidOperation - 使用
resume_after或start_after记录上一次成功处理的_id（即change["_id"]），避免重复或丢失事件 - 不要在异常后直接
break，而应重建 stream：先 close 原 stream，再调用collection.watch(..., resume_after=last_id) - 加退避重试（如
time.sleep(1)），避免密集轮询打爆连接数
watch() 的关键参数怎么选：pipeline、full_document、max_await_time_ms
这些参数直接影响数据实时性、带宽消耗和内存占用。不设或乱设会导致漏事件、高延迟或 OOM。
实操建议：
立即学习“Python免费学习笔记（深入）”；
-
pipeline：用[{"$match": {"operationType": {"$in": ["insert", "update"]}}}]过滤，别在 Python 层后过滤——oplog 事件量大时，前置过滤能显著降低传输和反序列化开销 -
full_document="updateLookup"：仅在需要更新后完整文档时开启；它会额外查一次当前文档，增加延迟和读负载，生产环境慎用 -
max_await_time_ms：设为5000（5 秒）较稳妥；太小（如 100ms）易频繁唤醒，太大（如 30s）会导致事件积压感知延迟高 - 避免传
batch_size给watch()——它不生效，change stream 本身不支持批量拉取
聚合流（aggregate + $changeStream）与普通 watch() 的区别和适用场景
PyMongo 的 collection.watch() 底层就是封装了 aggregate([{"$changeStream": {...}}])。但直接调用 aggregate() 可以混用其他 stage（如 $lookup、$project），适合复杂转换逻辑；而 watch() 更简洁、类型安全、自动处理 resumable 逻辑。
实操建议：
立即学习“Python免费学习笔记（深入）”；
- 优先用
watch()：它自动注入resumeToken处理、兼容未来 MongoDB 版本变更、API 更稳定 - 只有当你明确需要在服务端做
$lookup关联另一集合，且能接受无法自动 resume（需手动管理startAtOperationTime）时，才用原生aggregate()+$changeStream - 注意：
$changeStreamstage 必须是 pipeline 第一个 stage，且不能跟在$facet或$unionWith后面 - 直接调用
db.command("aggregate", ...)会绕过 PyMongo 的 cursor 自动重连机制，务必自行实现错误恢复
真正难的不是启动 stream，而是让它的生命周期与业务逻辑解耦、状态可持久化、重启后能精准续读——token 存哪、失败时日志够不够定位、是否要跨进程同步 last_id，这些才是线上落地时最常卡住的地方。
