3. 搭建可观测的变更流实验环境
Change Streams 依赖副本集或分片集群的 oplog 机制，单节点实例无法开启 watch()。因此实验环境的第一步不是写代码，而是把 MongoDB 部署成具备复制能力的拓扑，并让审计、日志、监控三类可观测能力同时就位。本节以“单节点副本集 + 审计日志 + 监控采集”为主线，给出可直接复现的搭建过程，并验证 watch() 的基本行为。
一、拓扑选择：单节点副本集还是三节点副本集
Change Streams 要求底层是副本集或分片集群。单节点副本集（rs.initiate 时只列一个成员）同样能产生 oplog，因此可以开启 Change Streams，适合本地实验；但它没有冗余，节点重启期间 oplog 会中断，不适合验证故障切换场景。三节点副本集（1 主 2 从）能覆盖主从切换、oplog 追赶、readConcern 与 writeConcern 组合等更真实的行为。
建议实验环境采用三节点副本集，端口规划为 27017、27018、27019，数据目录分别为 /data/rs0-0、/data/rs0-1、/data/rs0-2。若资源受限，可先用单节点副本集跑通 watch()，再升级为三节点。分片集群（mongos + config server + shard）在验证跨分片 Change Streams 时才需要，本节先不展开。
二、启动副本集并开启审计与日志
每个节点使用独立的配置文件启动。以下为 27017 节点的示例配置，重点开启审计日志、慢查询日志和 oplog 大小设置：
# /etc/mongod-rs0-0.conf
storage:
dbPath: /data/rs0-0
journal:
enabled: true
wiredTiger:
engineConfig:
cacheSizeGB: 1
systemLog:
destination: file
path: /var/log/mongodb/rs0-0.log
logAppend: true
verbosity: 1
net:
port: 27017
bindIp: 127.0.0.1
replication:
replSetName: rs0
oplogSizeMB: 512
auditLog:
destination: file
format: JSON
path: /var/log/mongodb/rs0-0-audit.json
filter: '{ atype: { $in: ["authenticate","createCollection","dropCollection","createIndex","dropIndex"] } }'
operationProfiling:
mode: slowOp
slowOpThresholdMs: 100
setParameter:
enableTestCommands: 1
审计日志的 filter 决定了记录范围。实验环境建议至少覆盖 authenticate、createCollection、dropCollection、createIndex、dropIndex 五类操作，这样在验证 watch() 时能同时看到 DDL 事件与审计记录的对应关系。生产环境应进一步加入 update、delete、insert 等写操作，但要注意审计日志体积增长。
三个节点分别用 mongod -f 启动后，连接 27017 执行初始化：
rs.initiate({
_id: "rs0",
members: [
{ _id: 0, host: "127.0.0.1:27017" },
{ _id: 1, host: "127.0.0.1:27018" },
{ _id: 2, host: "127.0.0.1:27019" }
]
})
执行 rs.status() 确认 members 中有一个 PRIMARY、两个 SECONDARY，且每个成员的 optime 在持续前进。若某个节点状态为 STARTUP2 或 RECOVERING，说明数据同步尚未完成，此时不要开始 watch() 验证。
三、准备数据集与验证 watch() 基本行为
Change Streams 的验证需要一组能产生 insert、update、delete、replace 四类事件的数据。建议准备一个 orders 集合，字段包含 orderId、status、amount、updatedAt，并建立 status 索引，便于后续按状态过滤事件。
use shop
db.orders.createIndex({ status: 1 })
db.orders.insertMany([
{ orderId: "A1001", status: "created", amount: 199, updatedAt: new Date() },
{ orderId: "A1002", status: "created", amount: 88, updatedAt: new Date() }
])
在 mongosh 中开启 watch()，观察事件流：
const cs = db.orders.watch([], { fullDocument: "updateLookup" })
while (!cs.isClosed()) {
const ev = cs.tryNext()
if (ev) printjson({
op: ev.operationType,
id: ev.documentKey._id,
ns: ev.ns.coll,
doc: ev.fullDocument
})
}
另开一个 mongosh 会话执行写操作：插入 A1003、把 A1001 的 status 改为 paid、删除 A1002。观察第一个会话是否依次收到 insert、update、delete 三类事件。判断依据是：operationType 与写操作一致，documentKey._id 与目标文档一致，update 事件在 fullDocument: "updateLookup" 下能拿到更新后的完整文档。
若 watch() 没有输出，按以下顺序排查：确认连接的是副本集而非单节点；确认当前用户对 shop.orders 有 find 与 changeStream 权限；确认 oplog 未被写满覆盖（rs.printReplicationInfo() 查看 oplog 窗口）；确认没有在事务中开启 watch()（Change Streams 不支持事务内 watch）。
四、可观测性三件套的落地位置
审计日志回答“谁在什么时候对哪个集合做了什么”，系统日志回答“节点内部发生了什么”，监控指标回答“资源与复制状态是否健康”。三者缺一不可。建议在实验环境中同时开启：审计日志写入独立文件，系统日志保留 logAppend，监控侧用 mongostat 或 Prometheus + mongodb_exporter 采集 metrics.repl、metrics.oplog、metrics.document 等指标。
五、验证结果与常见偏差
一次完整的验证应得到如下结果：watch() 会话依次输出 insert（A1003）、update（A1001，fullDocument.status 为 paid）、delete（A1002）三条事件；审计日志中出现对应的 createIndex 与 insert 记录；rs.printReplicationInfo() 显示 oplog 窗口大于实验时长。若 update 事件的 fullDocument 为 null，说明未设置 fullDocument: "updateLookup"，或更新操作发生在 watch() 开启之前。
常见偏差还包括：事件顺序与写操作顺序不一致（多线程写入时正常，单线程写入时应一致）；delete 事件没有 fullDocument（这是预期行为，delete 只带 documentKey）；resumeToken 在每次事件后变化，可用于断点续传。建议在实验记录中保存第一条事件的 resumeToken，后续用 resumeAfter 验证断点恢复能力。