MongoDB 变更流类型化监听

示例说明

轮询集合 diff 既慢又易漏，变更流直接订阅数据库的写入日志：插入、更新、删除事件实时推送。管道首段的 $match 在服务端过滤，只传关注的事件类型；updateLookup 让更新事件附带完整文档，省去二次查询。resumeToken 是事件流的位点书签——进程重启后从令牌处继续，不重放也不漏掉中间变更。

技术要点

• $match 服务端过滤事件类型
• fullDocument 模式补全更新事件载荷
• resumeToken 支撑断点续传

代码解读

集合泛型让事件里的 fullDocument 带 OrderDoc 形态；监听器按操作类型分支处理并持续更新续传令牌。插入一条订单演示事件触达；错误监听负责异常时关闭流，超时收尾同时关闭流与客户端连接释放资源。

使用建议

变更流需要副本集或分片集群，单机不支持。令牌要持久化到可靠存储才能跨进程重启。事件处理要幂等，断线重连可能重放窗口内事件。

代码示例

import { MongoClient, ChangeStreamDocument } from 'mongodb';

interface OrderDoc {
  _id: object;
  status: 'pending' | 'shipped' | 'delivered';
  total: number;
}

const client = new MongoClient(process.env.MONGO_URI ?? 'mongodb://127.0.0.1:27017');
await client.connect();

const orders = client.db('shop').collection<OrderDoc>('orders');

let resumeToken: string | null = null;

const stream = orders.watch(
  [{ $match: { operationType: { $in: ['insert', 'update'] } } }],
  { fullDocument: 'updateLookup' },
);

stream.on('change', (event: ChangeStreamDocument<OrderDoc>) => {
  if (event.operationType === 'insert' || event.operationType === 'update') {
    const doc = event.fullDocument;
    console.log('order changed:', doc?._id, 'status', doc?.status);
  }
  // 断点续传：记录最新令牌，重启后从该位置继续
  resumeToken = JSON.stringify(stream.resumeToken);
});

stream.on('error', async (err) => {
  console.error('stream failed:', err.message);
  await stream.close().catch(() => {});
});

await orders.insertOne({ status: 'pending', total: 1200 } as OrderDoc);

setTimeout(() => stream.close().then(() => client.close()), 2000);