如何保持MongoDB Change Stream持续运行实现常驻数据监听？

你现在的代码执行完就退出，根本不是需要选asyncio还是多线程的问题——你只调用了一次next(stream)取到第一条事件，没有做持续迭代，逻辑自然跑完就结束了。

绝大多数场景下根本不需要搞异步/多线程，同步无限循环是稳定性最高、坑最少的方案，也是官方推荐的常驻监听写法，自带断点续传、异常重连能力：

import time
from pymongo.errors import PyMongoError, ConnectionError

db = pymongo_util.get_collection("DataDB","XYZ_Collection")
# 过滤只监听插入事件，不需要可以删除pipeline配置，减少无效事件传输
pipeline = [{'$match': {'operationType': 'insert'}}]
resume_token = None  # 存储消费断点，重启/断连后不丢事件

while True:
    try:
        # 传入resume_after实现断点续传
        with db.watch(
            pipeline=pipeline,
            full_document="updateLookup",
            resume_after=resume_token
        ) as stream:
            for document in stream:
                resume_token = stream.resume_token
                # 此处编写新文档的自定义处理逻辑
                print("捕获到新增文档：", document['fullDocument'])
                # process_new_doc(document['fullDocument'])
    except (ConnectionError, PyMongoError) as e:
        print(f"监听连接异常，5秒后重连: {str(e)}")
        time.sleep(5)
        continue

代码逻辑说明：

• 直接对watch()返回的stream对象做for循环迭代即可，迭代过程会自动阻塞等待新事件，不会空转占CPU，不需要手动调用next()
• 维护resume_token是为了断连、进程重启后能从上次中断的位置继续消费，不会漏掉断连期间产生的新文档
• 捕获连接相关异常做休眠重连，应对网络闪断、MongoDB主从切换、临时运维重启等场景

• 异步服务同进程部署：如果要和FastAPI这类异步Web服务同进程运行，不要用同步pymongo，换成异步驱动motor，监听逻辑和同步版完全一致，只是把普通for循环换成async for即可，避免阻塞整个异步事件循环
• 同步服务同进程部署：如果要和Flask这类同步Web服务同进程运行，单独开一个守护线程跑上面的同步监听逻辑即可，不要把监听逻辑放到Web请求的生命周期里，避免被Web服务的进程回收机制打断
• 生产最优部署方式：更建议把监听器作为独立的守护进程部署，和Web服务完全隔离，互不影响稳定性

• Change Stream功能要求MongoDB部署形态为副本集或者分片集群，单节点实例无法使用该功能
• 监听使用的数据库账号需要拥有目标集合的changeStream和find权限，否则会无报错退出
• 不需要强行追求非阻塞：同步监听的消费性能足够覆盖绝大多数业务场景，额外引入异步/多线程只会增加故障排查成本

内容的提问来源于stack exchange，提问作者keynesiancross