如何在Python微服务中利用MongoDB的Capped Collection实现事件广播？-MongoDB-PHP中文网

Capped Collection是MongoDB中预分配固定大小、自动覆盖最旧文档的环形缓冲区式集合，适合事件广播因其支持tailable cursor实现低延迟实时监听、高写入吞吐、无索引开销且免维护滚动清理。

如何在python微服务中利用mongodb的capped collection实现事件广播？

什么是Capped Collection，它为什么适合事件广播？

Capped Collection 是 MongoDB 中一种固定大小的集合，写入时自动按插入顺序覆盖最老文档。它天然支持 tailable cursor —— 一种可“挂起等待新数据”的游标，这正是事件广播的核心机制：服务启动后监听集合末尾，新事件一写入就立刻被拉取。

它比普通集合更适合广播场景，因为：

不需要手动维护游标位置（如 _id 或时间戳），tailable cursor 自动跟踪
无索引开销（默认只有 _id 索引），写入吞吐高
滚动覆盖避免无限增长，省去定时清理逻辑

但要注意：Capped Collection 不支持 update、delete 和任意排序，只适合“只追加 + 只读消费”的事件流。

如何创建并正确配置 Capped Collection？

创建必须在集合首次写入前完成，且不能通过 ORM（如 MongoEngine）直接声明，得用 PyMongo 原生命令：

from pymongo import MongoClient
<p>client = MongoClient("mongodb://localhost:27017/")
db = client["eventbus"]</p><h1>必须显式创建，size 单位是字节，max 是文档数（二者选一，size 优先）</h1><p>db.create_collection(
"events",
capped=True,
size=1024 <em> 1024 </em> 50,  # 50MB
max=10000               # 或最多 10000 条（超出则按时间覆盖）
)</p>

关键点：

size 和 max 不能同时为 0；生产环境建议以 size 为主，避免小文档导致过早截断
创建后不可修改为非 capped，也不可修改 size/max
集合名不要带特殊字符或空格，否则 tailable cursor 在某些驱动版本中会静默失败

如何用 tailable cursor 实现可靠事件监听？

PyMongo 的 find 需配合 cursor_type=CursorType.TAILABLE_AWAIT 和合理超时，否则容易断连或漏事件：

from pymongo import CursorType
import time
<p>coll = db["events"]
cursor = coll.find(
{},  # 通常不加过滤，广播所有事件
cursor_type=CursorType.TAILABLE_AWAIT,
oplog_replay=True,  # 兼容旧版驱动，新驱动可省略
no_cursor_timeout=True  # 防止服务器端游标过期
)</p><p>while cursor.alive:
try:
doc = next(cursor)
print("收到事件:", doc)</p><h1>处理事件（注意：此处不应阻塞太久，否则游标可能超时）</h1><pre class="brush:php;toolbar:false;">except StopIteration:
    # 暂无新数据，await 期间可能被唤醒
    time.sleep(0.1)
except Exception as e:
    print("游标异常:", e)
    break  # 实际中应重连并重新获取 cursor

Secure Python Code Sandbox

安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。

下载

常见陷阱：

忘记设 no_cursor_timeout=True，导致游标几秒后自动关闭，监听中断
在 next(cursor) 外层没包 try/except，一次网络抖动就退出整个循环
用 find().sort(...) —— Capped Collection 不支持除 $natural 外的排序，会报错 CommandNotSupportedOnCapped