MongoDB Change Streams：实时监听数据变化的实战场景

第一章：Change Streams 核心概念与架构原理

1.1 Change Streams 技术本质

MongoDB Change Streams 是 MongoDB 3.6 版本引入的核心功能，它提供了基于发布-订阅模式的实时数据变更监听机制。与传统的轮询查询或操作日志（oplog）直接访问不同，Change Streams 提供了更高级、更安全的API来监听数据库的实时变化。
技术原理深度解析：
Change Streams 底层基于 MongoDB 的复制机制，但提供了完全不同的抽象层级：

1. 基于Oplog的封装：Change Streams 本质上是对 oplog（操作日志）的高级封装，但避免了直接访问 oplog 的复杂性和风险。
2. 游标机制：使用 tailable cursor 在 oplog 集合上持续监听新操作。
3. 数据转换：将原始的 oplog 条目转换为更易理解的变更事件格式。
4. 恢复机制：内置 resume token 机制，确保连接中断后能够从断点恢复。
   与传统方案的对比：

1.2 Change Streams 事件模型架构

Change Streams 的架构设计采用了现代化的流处理模式，其核心组件和数据处理流程如下图所示：

这个架构展示了 Change Streams 如何作为 MongoDB 和外部系统之间的桥梁，将数据库的变更事件实时地传递到各种下游系统和应用中。

1.3 变更事件格式详解

Change Streams 产生的事件包含丰富的元数据和实际变更内容：

{
  "_id": {
    "_data": "8262B4042B0000000129295A1004...", // 恢复令牌
    "clusterTime": {
      "$timestamp": {
        "t": 1672531200,
        "i": 1
      }
    }
  },
  "operationType": "insert", // 操作类型
  "ns": {
    "db": "ecommerce",
    "coll": "orders"
  },
  "documentKey": {
    "_id": ObjectId("507f1f77bcf86cd799439011")
  },
  "fullDocument": {
    "_id": ObjectId("507f1f77bcf86cd799439011"),
    "customerId": "12345",
    "amount": 99.99,
    "status": "created"
  },
  "wallTime": ISODate("2023-01-01T00:00:00Z")
}

关键字段解析：

• _id：包含恢复令牌，用于断点续传
• operationType：操作类型（insert、update、replace、delete等）
• ns：命名空间（数据库和集合）
• documentKey：受影响文档的主键
• fullDocument：操作后的完整文档（仅insert和replace时可用）
• updateDescription：更新操作的详细描述（仅update时可用）

第二章：Change Streams 核心API与实战基础

2.1 基础监听模式与配置

基础监听示例：

// 创建Change Stream
const pipeline = [
  { $match: { operationType: { $in: ['insert', 'update', 'delete'] } } },
  { $project: { documentKey: 1, fullDocument: 1, operationType: 1 } }
];

const changeStream = db.collection('orders').watch(pipeline);

// 监听变更事件
changeStream.on('change', (change) => {
  console.log('收到变更事件:', change);
  switch (change.operationType) {
    case 'insert':
      handleOrderCreated(change.fullDocument);
      break;
    case 'update':
      handleOrderUpdated(change.documentKey._id, change.updateDescription);
      break;
    case 'delete':
      handleOrderDeleted(change.documentKey._id);
      break;
  }
});

// 错误处理
changeStream.on('error', (error) => {
  console.error('Change Stream错误:', error);
  // 实现重连逻辑
  reconnectChangeStream();
});

高级配置选项：

const changeStream = db.collection('orders').watch([], {
  fullDocument: 'updateLookup', // 获取更新后的完整文档
  resumeAfter: resumeToken,     // 从特定断点恢复
  maxAwaitTimeMs: 1000,         // 等待新事件的最大时间
  batchSize: 100,               // 每批返回的最大事件数
  startAtOperationTime: timestamp, // 从特定时间开始
  collation: { locale: 'en', strength: 2 } // 排序规则
});

2.2 恢复令牌与断点续传机制

恢复令牌管理：

class ChangeStreamManager {
  constructor(collection, storagePath) {
    this.collection = collection;
    this.storagePath = storagePath;
    this.currentResumeToken = null;
    this.changeStream = null;
  }
  
  // 启动Change Stream
  async start() {
    // 尝试加载之前的恢复令牌
    const savedToken = await this.loadResumeToken();
    
    const options = {
      fullDocument: 'updateLookup'
    };
    
    if (savedToken) {
      options.resumeAfter = savedToken;
      console.log('从断点恢复:', savedToken);
    }
    
    this.changeStream = this.collection.watch([], options);
    
    this.changeStream.on('change', (change) => {
      this.currentResumeToken = change._id;
      this.handleChange(change);
    });
    
    this.changeStream.on('error', (error) => {
      console.error('Change Stream错误:', error);
      this.restart().catch(console.error);
    });
  }
  
  // 处理变更事件
  handleChange(change) {
    try {
      // 业务逻辑处理
      this.processBusinessEvent(change);
      
      // 定期保存恢复令牌（每10个事件）
      if (this.eventCount % 10 === 0) {
        this.saveResumeToken(this.currentResumeToken);
      }
      
      this.eventCount++;
    } catch (error) {
      console.error('处理变更事件错误:', error);
    }
  }
  
  // 保存恢复令牌
  async saveResumeToken(token) {
    try {
      await fs.writeFile(
        this.storagePath, 
        JSON.stringify(token)
      );
    } catch (error) {
      console.error('保存恢复令牌失败:', error);
    }
  }
  
  // 加载恢复令牌
  async loadResumeToken() {
    try {
      if (await fs.exists(this.storagePath)) {
        const data = await fs.readFile(this.storagePath, 'utf8');
        return JSON.parse(data);
      }
    } catch (error) {
      console.error('加载恢复令牌失败:', error);
    }
    return null;
  }
  
  // 重启Change Stream
  async restart() {
    if (this.changeStream) {
      this.changeStream.close();
    }
    await new Promise(resolve => setTimeout(resolve, 5000)); // 等待5秒
    await this.start();
  }
}

第三章：高级应用场景与实战模式

3.1 实时数据同步系统

多目标数据同步架构：

class DataSynchronizer {
  constructor(sourceCollection, targetClients) {
    this.sourceCollection = sourceCollection;
    this.targetClients = targetClients; // 多个目标数据库客户端
    this.changeStream = null;
  }
  
  async startSync() {
    const pipeline = [
      {
        $match: {
          operationType: { $in: ['insert', 'update', 'replace', 'delete'] }
        }
      }
    ];
    
    this.changeStream = this.sourceCollection.watch(pipeline, {
      fullDocument: 'updateLookup'
    });
    
    this.changeStream.on('change', async (change) => {
      try {
        await this.syncToTargets(change);
      } catch (error) {
        console.error('同步失败:', error);
        // 实现重试机制
        await this.retrySync(change);
      }
    });
  }
  
  async syncToTargets(change) {
    const operations = this.targetClients.map(client => 
      this.applyChangeToTarget(client, change)
    );
    
    // 并行执行所有同步操作
    await Promise.all(operations);
  }
  
  async applyChangeToTarget(client, change) {
    const targetCollection = client.db('targetDb').collection('targetColl');
    
    switch (change.operationType) {
      case 'insert':
        await targetCollection.insertOne(change.fullDocument);
        break;
      case 'update':
        await targetCollection.updateOne(
          { _id: change.documentKey._id },
          { $set: change.updateDescription.updatedFields },
          { upsert: true }
        );
        break;
      case 'replace':
        await targetCollection.replaceOne(
          { _id: change.documentKey._id },
          change.fullDocument,
          { upsert: true }
        );
        break;
      case 'delete':
        await targetCollection.deleteOne({ _id: change.documentKey._id });
        break;
    }
  }
  
  async retrySync(change, maxRetries = 3) {
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        await this.syncToTargets(change);
        return; // 成功则退出
      } catch (error) {
        if (attempt === maxRetries) {
          throw new Error(`同步失败 after ${maxRetries} 次重试: ${error.message}`);
        }
        await new Promise(resolve => 
          setTimeout(resolve, 1000 * Math.pow(2, attempt)) // 指数退避
        );
      }
    }
  }
}

3.2 复杂事件处理与实时分析

实时订单分析系统：

class RealTimeOrderAnalytics {
  constructor(orderCollection) {
    this.orderCollection = orderCollection;
    this.orderStats = {
      totalOrders: 0,
      totalRevenue: 0,
      hourlyRevenue: new Map(),
      categoryRevenue: new Map()
    };
  }
  
  startAnalytics() {
    const pipeline = [
      {
        $match: {
          operationType: 'insert',
          'fullDocument.status': 'completed'
        }
      },
      {
        $project: {
          order: '$fullDocument',
          operationType: 1
        }
      }
    ];
    
    const changeStream = this.orderCollection.watch(pipeline);
    
    changeStream.on('change', (change) => {
      this.updateOrderStats(change.order);
      this.detectAnomalies(change.order);
      this.updateRealTimeDashboard();
    });
  }
  
  updateOrderStats(order) {
    const orderAmount = order.amount || 0;
    const orderHour = new Date(order.createdAt).getHours();
    const category = order.category || 'unknown';
    
    // 更新统计信息
    this.orderStats.totalOrders++;
    this.orderStats.totalRevenue += orderAmount;
    
    // 小时级统计
    const hourlyRevenue = this.orderStats.hourlyRevenue.get(orderHour) || 0;
    this.orderStats.hourlyRevenue.set(orderHour, hourlyRevenue + orderAmount);
    
    // 分类统计
    const categoryRevenue = this.orderStats.categoryRevenue.get(category) || 0;
    this.orderStats.categoryRevenue.set(category, categoryRevenue + orderAmount);
    
    // 清理旧数据（保留24小时）
    if (this.orderStats.hourlyRevenue.size > 24) {
      const currentHour = new Date().getHours();
      for (const [hour] of this.orderStats.hourlyRevenue) {
        if (Math.abs(hour - currentHour) > 24) {
          this.orderStats.hourlyRevenue.delete(hour);
        }
      }
    }
  }
  
  detectAnomalies(order) {
    // 异常检测逻辑
    const currentHour = new Date().getHours();
    const hourlyAvg = this.calculateHourlyAverage(currentHour);
    
    if (order.amount > hourlyAvg * 3) {
      // 检测到大额订单
      this.triggerAlert('large_order', {
        orderId: order._id,
        amount: order.amount,
        expectedMax: hourlyAvg * 3
      });
    }
    
    // 频率检测
    if (this.orderStats.totalOrders > 1000) {
      const ordersLastMinute = this.getOrdersInTimeWindow(60 * 1000);
      if (ordersLastMinute > 100) {
        this.triggerAlert('high_frequency', {
          ordersPerMinute: ordersLastMinute
        });
      }
    }
  }
  
  calculateHourlyAverage(hour) {
    const hourlyData = this.orderStats.hourlyRevenue;
    if (hourlyData.size === 0) return 0;
    
    let total = 0;
    let count = 0;
    
    for (const [h, revenue] of hourlyData) {
      if (Math.abs(h - hour) <= 6) { // 考虑相近时间段
        total += revenue;
        count++;
      }
    }
    
    return count > 0 ? total / count : 0;
  }
}

第四章：生产环境部署与优化策略

4.1 集群环境下的Change Streams

分片集群配置：

// 分片集群下的Change Streams配置
const mongoose = require('mongoose');
const { ReplSet, ShardedCluster } = require('mongodb-topology-manager');

class ShardedChangeStreamManager {
  constructor(mongoUri, shardConfig) {
    this.mongoUri = mongoUri;
    this.shardConfig = shardConfig;
    this.changeStreams = new Map();
  }
  
  async initialize() {
    // 连接到mongos路由器
    this.client = await mongoose.createConnection(this.mongoUri, {
      useNewUrlParser: true,
      useUnifiedTopology: true,
      readPreference: 'secondaryPreferred',
      maxPoolSize: 50,
      minPoolSize: 10
    });
    
    // 为每个分片创建独立的Change Stream
    for (const shard of this.shardConfig.shards) {
      await this.createShardChangeStream(shard);
    }
  }
  
  async createShardChangeStream(shard) {
    try {
      const shardDb = this.client.db(shard.database);
      const pipeline = [
        {
          $match: {
            operationType: { $in: ['insert', 'update', 'delete'] },
            'ns.db': shard.database
          }
        }
      ];
      
      const changeStream = shardDb.watch(pipeline, {
        fullDocument: 'updateLookup',
        batchSize: 100,
        maxAwaitTimeMs: 1000
      });
      
      changeStream.on('change', (change) => {
        this.handleShardChange(shard.name, change);
      });
      
      changeStream.on('error', (error) => {
        console.error(`分片 ${shard.name} Change Stream错误:`, error);
        this.recoverShardStream(shard);
      });
      
      this.changeStreams.set(shard.name, changeStream);
      
    } catch (error) {
      console.error(`创建分片 ${shard.name} Change Stream失败:`, error);
    }
  }
  
  handleShardChange(shardName, change) {
    // 根据分片路由处理变更事件
    const eventKey = `${shardName}_${change.ns.coll}_${change.operationType}`;
    this.eventProcessor.process(eventKey, change);
    
    // 更新监控指标
    this.metrics.increment(`changes.${shardName}.${change.operationType}`);
  }
  
  async recoverShardStream(shard) {
    console.log(`尝试恢复分片 ${shard.name} 的Change Stream...`);
    
    // 关闭现有的Change Stream
    const oldStream = this.changeStreams.get(shard.name);
    if (oldStream) {
      oldStream.close();
    }
    
    // 等待一段时间后重试
    await new Promise(resolve => setTimeout(resolve, 5000));
    
    try {
      await this.createShardChangeStream(shard);
      console.log(`分片 ${shard.name} Change Stream恢复成功`);
    } catch (error) {
      console.error(`分片 ${shard.name} Change Stream恢复失败:`, error);
      // 加入重试队列
      this.retryQueue.add(shard);
    }
  }
}

4.2 性能优化与资源管理

资源优化配置：

# Docker容器资源限制
version: '3.8'
services:
  change-stream-processor:
    image: node:18
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: '2'
        reservations:
          memory: 1G
          cpus: '0.5'
    environment:
      - NODE_OPTIONS=--max-old-space-size=1536
      - UV_THREADPOOL_SIZE=16
    volumes:
      - ./app:/app
    working_dir: /app

# Kubernetes部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: change-stream-processor
spec:
  replicas: 3
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 1
  template:
    spec:
      containers:
      - name: processor
        image: change-stream-processor:latest
        resources:
          limits:
            memory: "2Gi"
            cpu: "2000m"
          requests:
            memory: "1Gi"
            cpu: "500m"
        env:
        - name: NODE_OPTIONS
          value: "--max-old-space-size=1536"
        - name: UV_THREADPOOL_SIZE
          value: "16"
        livenessProbe:
          httpGet:
            path: /health
            port: 3000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 3000
          initialDelaySeconds: 5
          periodSeconds: 5

性能监控与调优：

class ChangeStreamMonitor {
  constructor() {
    this.metrics = {
      eventsProcessed: 0,
      eventsPerSecond: 0,
      averageProcessingTime: 0,
      errorCount: 0,
      memoryUsage: 0
    };
    
    this.startTime = Date.now();
    this.eventTimestamps = [];
  }
  
  recordEventProcessing(startTime) {
    const processingTime = Date.now() - startTime;
    
    this.metrics.eventsProcessed++;
    this.eventTimestamps.push(Date.now());
    
    // 计算每秒事件数（滑动窗口）
    const now = Date.now();
    const windowStart = now - 1000;
    this.eventTimestamps = this.eventTimestamps.filter(ts => ts > windowStart);
    this.metrics.eventsPerSecond = this.eventTimestamps.length;
    
    // 计算平均处理时间（指数移动平均）
    this.metrics.averageProcessingTime = 
      this.metrics.averageProcessingTime * 0.9 + processingTime * 0.1;
    
    // 记录内存使用
    this.metrics.memoryUsage = process.memoryUsage().heapUsed / 1024 / 1024;
    
    // 检查性能异常
    this.checkPerformanceAnomalies();
  }
  
  checkPerformanceAnomalies() {
    // 事件积压检测
    if (this.metrics.eventsPerSecond > 1000 && 
        this.metrics.averageProcessingTime > 100) {
      this.triggerAlert('high_backpressure', {
        eventsPerSecond: this.metrics.eventsPerSecond,
        avgProcessingTime: this.metrics.averageProcessingTime
      });
    }
    
    // 内存泄漏检测
    if (this.metrics.memoryUsage > 1024) { // 超过1GB
      this.triggerAlert('high_memory_usage', {
        memoryUsage: this.metrics.memoryUsage
      });
    }
  }
  
  getMetrics() {
    const uptime = (Date.now() - this.startTime) / 1000;
    return {
      ...this.metrics,
      uptime: uptime,
      eventsPerMinute: this.metrics.eventsProcessed / (uptime / 60)
    };
  }
  
  async exportMetrics() {
    const metrics = this.getMetrics();
    
    // 推送到Prometheus
    await this.pushToPrometheus(metrics);
    
    // 记录到日志
    console.log('性能指标:', JSON.stringify(metrics));
    
    // 发送到监控系统
    await this.sendToMonitoringSystem(metrics);
  }
}

第五章：安全性与可靠性保障

5.1 安全认证与授权

安全的Change Streams配置：

const { MongoClient } = require('mongodb');

class SecureChangeStreamClient {
  constructor(config) {
    this.config = config;
    this.client = null;
    this.changeStream = null;
  }
  
  async connect() {
    const connectionOptions = {
      useNewUrlParser: true,
      useUnifiedTopology: true,
      ssl: this.config.useSSL,
      sslValidate: this.config.sslValidate,
      sslCA: this.config.sslCA ? fs.readFileSync(this.config.sslCA) : null,
      sslCert: this.config.sslCert ? fs.readFileSync(this.config.sslCert) : null,
      sslKey: this.config.sslKey ? fs.readFileSync(this.config.sslKey) : null,
      authMechanism: 'SCRAM-SHA-256',
      authSource: 'admin',
      readPreference: 'secondary',
      w: 'majority',
      j: true,
      wtimeout: 10000
    };
    
    this.client = new MongoClient(this.config.connectionString, connectionOptions);
    
    await this.client.connect();
    
    // 验证连接权限
    await this.validatePermissions();
    
    return this.client;
  }
  
  async validatePermissions() {
    const adminDb = this.client.db('admin');
    
    try {
      // 检查change streams权限
      const result = await adminDb.command({
        listCollections: 1,
        filter: { name: 'system.views' }
      });
      
      // 检查具体集合的读权限
      const testRead = await this.client.db(this.config.database)
        .collection(this.config.collection)
        .findOne({}, { projection: { _id: 1 } });
      
      console.log('权限验证通过');
      
    } catch (error) {
      throw new Error(`权限验证失败: ${error.message}`);
    }
  }
  
  async startSecureChangeStream() {
    const pipeline = [
      {
        $match: {
          operationType: { $in: ['insert', 'update', 'delete'] }
        }
      },
      {
        $redact: {
          $cond: {
            if: {
              $or: [
                { $eq: ['$operationType', 'delete'] },
                { 
                  $and: [
                    { $eq: ['$operationType', 'update'] },
                    { $gt: ['$fullDocument.sensitive', false] }
                  ]
                }
              ]
            },
            then: '$$PRUNE',
            else: '$$KEEP'
          }
        }
      }
    ];
    
    this.changeStream = this.client.db(this.config.database)
      .collection(this.config.collection)
      .watch(pipeline, {
        fullDocument: 'updateLookup',
        maxAwaitTimeMs: 1000,
        batchSize: 50
      });
    
    this.changeStream.on('change', (change) => {
      this.handleSecureChange(change);
    });
    
    this.changeStream.on('error', (error) => {
      this.handleSecureError(error);
    });
  }
  
  handleSecureChange(change) {
    // 数据脱敏处理
    const sanitizedChange = this.sanitizeData(change);
    
    // 审计日志记录
    this.logAuditEvent(sanitizedChange);
    
    // 业务处理
    this.processBusinessLogic(sanitizedChange);
  }
  
  sanitizeData(change) {
    // 移除敏感字段
    if (change.fullDocument) {
      const { password, creditCard, ssn, ...safeDocument } = change.fullDocument;
      change.fullDocument = safeDocument;
    }
    
    if (change.updateDescription && change.updateDescription.updatedFields) {
      const { password, creditCard, ssn, ...safeFields } = change.updateDescription.updatedFields;
      change.updateDescription.updatedFields = safeFields;
    }
    
    return change;
  }
  
  async logAuditEvent(change) {
    const auditDb = this.client.db('audit');
    await auditDb.collection('change_events').insertOne({
      timestamp: new Date(),
      operation: change.operationType,
      namespace: `${change.ns.db}.${change.ns.coll}`,
      documentKey: change.documentKey,
      user: this.config.username,
      clientIp: this.config.clientIp,
      changeData: change
    });
  }
}

5.2 容错与灾难恢复

高可用Change Streams架构：

class HighAvailabilityChangeStream {
  constructor(primaryUri, replicaUris, options = {}) {
    this.primaryUri = primaryUri;
    this.replicaUris = replicaUris;
    this.options = options;
    this.activeClient = null;
    this.backupClients = new Map();
    this.currentResumeToken = null;
  }
  
  async initialize() {
    // 连接主集群
    try {
      await this.connectToPrimary();
    } catch (error) {
      console.warn('主集群连接失败，尝试备用集群:', error);
      await this.failoverToReplica();
    }
    
    // 连接备用集群
    await this.connectToReplicas();
    
    // 启动健康检查
    this.startHealthCheck();
  }
  
  async connectToPrimary() {
    this.activeClient = await this.createClient(this.primaryUri);
    await this.startChangeStream(this.activeClient);
  }
  
  async connectToReplicas() {
    for (const [id, uri] of this.replicaUris.entries()) {
      try {
        const client = await this.createClient(uri);
        this.backupClients.set(id, client);
        
        // 启动备份Change Stream（但不处理事件）
        await this.startBackupStream(client);
        
      } catch (error) {
        console.error(`备用集群 ${id} 连接失败:`, error);
      }
    }
  }
  
  async startChangeStream(client) {
    const changeStream = client.db(this.options.database)
      .collection(this.options.collection)
      .watch([], {
        resumeAfter: this.currentResumeToken,
        fullDocument: this.options.fullDocument || 'updateLookup'
      });
    
    changeStream.on('change', (change) => {
      this.currentResumeToken = change._id;
      this.handleChangeEvent(change);
    });
    
    changeStream.on('error', async (error) => {
      console.error('Change Stream错误:', error);
      await this.handleStreamError(error);
    });
    
    this.activeChangeStream = changeStream;
  }
  
  async handleStreamError(error) {
    // 检查错误类型
    if (this.isNetworkError(error) || this.isClusterError(error)) {
      console.log('检测到集群级别错误，开始故障转移...');
      await this.performFailover();
    } else if (this.isResumeError(error)) {
      console.log('恢复令牌失效，重新启动Change Stream...');
      this.currentResumeToken = null;
      await this.restartChangeStream();
    } else {
      console.log('未知错误，尝试重启...');
      await this.restartChangeStream();
    }
  }
  
  async performFailover() {
    // 关闭当前连接
    if (this.activeClient) {
      await this.activeClient.close();
    }
    
    // 尝试切换到备用集群
    for (const [id, client] of this.backupClients.entries()) {
      try {
        // 验证备用集群状态
        await this.validateReplicaClient(client);
        
        console.log(`切换到备用集群: ${id}`);
        this.activeClient = client;
        await this.startChangeStream(client);
        
        // 从新的备用集群列表移除当前使用的
        this.backupClients.delete(id);
        
        return; // 成功切换
        
      } catch (error) {
        console.error(`备用集群 ${id} 不可用:`, error);
      }
    }
    
    throw new Error('所有备用集群均不可用');
  }
  
  async validateReplicaClient(client) {
    // 检查集群状态
    const status = await client.db('admin').command({ replSetGetStatus: 1 });
    
    if (!status || !status.members) {
      throw new Error('无效的副本集状态');
    }
    
    // 检查是否有健康的主节点
    const primary = status.members.find(m => m.stateStr === 'PRIMARY');
    if (!primary) {
      throw new Error('没有可用的主节点');
    }
    
    // 检查复制延迟
    if (primary.optime && status.members[0].optime) {
      const lag = primary.optime.ts - status.members[0].optime.ts;
      if (lag > this.options.maxReplicationLag || 30000) {
        throw new Error(`复制延迟过高: ${lag}ms`);
      }
    }
  }
  
  startHealthCheck() {
    setInterval(async () => {
      try {
        await this.checkClusterHealth();
      } catch (error) {
        console.error('健康检查失败:', error);
      }
    }, this.options.healthCheckInterval || 30000);
  }
  
  async checkClusterHealth() {
    if (!this.activeClient) {
      throw new Error('没有活动的客户端连接');
    }
    
    // 检查连接状态
    await this.activeClient.db('admin').command({ ping: 1 });
    
    // 检查Change Stream状态
    if (!this.activeChangeStream) {
      throw new Error('Change Stream未运行');
    }
    
    // 检查事件处理延迟
    const lastEventTime = this.lastEventTimestamp;
    if (lastEventTime && Date.now() - lastEventTime > 60000) {
      throw new Error('事件处理延迟超过60秒');
    }
  }
}

第六章：监控、调试与性能优化

6.1 高级监控与警报系统

综合监控解决方案：

class ChangeStreamMonitor {
  constructor(options = {}) {
    this.options = {
      checkInterval: options.checkInterval || 30000,
      maxEventLag: options.maxEventLag || 60000,
      maxMemoryUsage: options.maxMemoryUsage || 1024,
      prometheusEnabled: options.prometheusEnabled || false
    };
    
    this.metrics = {
      eventsProcessed: 0,
      eventsPerSecond: 0,
      averageLag: 0,
      memoryUsage: 0,
      errorCount: 0,
      lastEventTime: null
    };
    
    this.historicalData = [];
    this.alertManager = new AlertManager();
  }
  
  startMonitoring() {
    // 定期收集指标
    this.monitoringInterval = setInterval(() => {
      this.collectMetrics();
      this.checkThresholds();
      this.exportMetrics();
    }, this.options.checkInterval);
    
    // 监听进程事件
    process.on('SIGTERM', () => this.stopMonitoring());
    process.on('SIGINT', () => this.stopMonitoring());
  }
  
  async collectMetrics() {
    const currentMetrics = {
      timestamp: Date.now(),
      eventsProcessed: this.metrics.eventsProcessed,
      eventsPerSecond: this.calculateEventsPerSecond(),
      averageLag: this.calculateAverageLag(),
      memoryUsage: process.memoryUsage().heapUsed / 1024 / 1024,
      errorCount: this.metrics.errorCount,
      uptime: process.uptime()
    };
    
    this.historicalData.push(currentMetrics);
    
    // 保持历史数据大小
    if (this.historicalData.length > 3600) { // 保留1小时数据（每30秒一次）
      this.historicalData.shift();
    }
    
    this.metrics = currentMetrics;
  }
  
  calculateEventsPerSecond() {
    const now = Date.now();
    const windowStart = now - 1000;
    const eventsInWindow = this.eventTimestamps.filter(ts => ts > windowStart).length;
    return eventsInWindow;
  }
  
  calculateAverageLag() {
    if (!this.lastEventTime) return 0;
    return Date.now() - this.lastEventTime;
  }
  
  checkThresholds() {
    // 检查事件积压
    if (this.metrics.averageLag > this.options.maxEventLag) {
      this.alertManager.triggerAlert('high_event_lag', {
        currentLag: this.metrics.averageLag,
        maxAllowed: this.options.maxEventLag
      });
    }
    
    // 检查内存使用
    if (this.metrics.memoryUsage > this.options.maxMemoryUsage) {
      this.alertManager.triggerAlert('high_memory_usage', {
        currentUsage: this.metrics.memoryUsage,
        maxAllowed: this.options.maxMemoryUsage
      });
    }
    
    // 检查错误率
    if (this.metrics.errorCount > 10 && 
        this.metrics.eventsProcessed > 100) {
      const errorRate = this.metrics.errorCount / this.metrics.eventsProcessed;
      if (errorRate > 0.1) { // 10%错误率
        this.alertManager.triggerAlert('high_error_rate', {
          errorRate: errorRate,
          errorCount: this.metrics.errorCount,
          totalEvents: this.metrics.eventsProcessed
        });
      }
    }
  }
  
  async exportMetrics() {
    if (this.options.prometheusEnabled) {
      await this.pushToPrometheus();
    }
    
    // 输出到日志
    console.log(JSON.stringify({
      type: 'metrics',
      timestamp: new Date().toISOString(),
      metrics: this.metrics
    }));
    
    // 发送到监控系统
    if (this.options.monitoringEndpoint) {
      await this.sendToMonitoringSystem();
    }
  }
  
  recordEventProcessing(startTime, success = true) {
    const processingTime = Date.now() - startTime;
    
    this.eventTimestamps.push(Date.now());
    this.metrics.eventsProcessed++;
    this.lastEventTime = Date.now();
    
    if (!success) {
      this.metrics.errorCount++;
    }
    
    // 更新性能指标
    this.metrics.averageProcessingTime = 
      (this.metrics.averageProcessingTime * 0.9) + (processingTime * 0.1);
  }
  
  stopMonitoring() {
    if (this.monitoringInterval) {
      clearInterval(this.monitoringInterval);
    }
    
    // 输出最终指标
    this.exportMetrics();
  }
}

6.2 调试与故障诊断

高级调试工具：

class ChangeStreamDebugger {
  constructor(changeStream, options = {}) {
    this.changeStream = changeStream;
    this.options = {
      logLevel: options.logLevel || 'info',
      captureEvents: options.captureEvents || false,
      maxCapturedEvents: options.maxCapturedEvents || 1000
    };
    
    this.capturedEvents = [];
    this.debugHandlers = new Map();
  }
  
  enableDebugging() {
    // 包装原始事件处理器
    const originalOn = this.changeStream.on.bind(this.changeStream);
    
    this.changeStream.on = (event, handler) => {
      if (event === 'change') {
        // 包装change事件处理器
        const wrappedHandler = (change) => {
          this.captureEvent(change);
          this.logEvent(change);
          this.notifyDebugHandlers(change);
          return handler(change);
        };
        
        return originalOn(event, wrappedHandler);
      }
      
      return originalOn(event, handler);
    };
    
    // 监听错误事件
    this.changeStream.on('error', (error) => {
      this.logError(error);
      this.captureError(error);
    });
  }
  
  captureEvent(change) {
    if (this.options.captureEvents) {
      this.capturedEvents.push({
        timestamp: Date.now(),
        event: change,
        operationType: change.operationType,
        namespace: `${change.ns.db}.${change.ns.coll}`
      });
      
      // 限制捕获的事件数量
      if (this.capturedEvents.length > this.options.maxCapturedEvents) {
        this.capturedEvents.shift();
      }
    }
  }
  
  logEvent(change) {
    const logLevel = this.options.logLevel;
    
    if (logLevel === 'debug') {
      console.debug('Change Stream事件:', {
        id: change._id,
        operation: change.operationType,
        ns: change.ns,
        documentKey: change.documentKey,
        wallTime: change.wallTime
      });
    } else if (logLevel === 'info') {
      console.info('Change Stream事件:', {
        operation: change.operationType,
        ns: `${change.ns.db}.${change.ns.coll}`,
        documentKey: change.documentKey
      });
    }
  }
  
  logError(error) {
    console.error('Change Stream错误:', {
      message: error.message,
      stack: error.stack,
      code: error.code,
      timestamp: new Date().toISOString()
    });
  }
  
  registerDebugHandler(name, handler) {
    this.debugHandlers.set(name, handler);
  }
  
  notifyDebugHandlers(change) {
    for (const [name, handler] of this.debugHandlers) {
      try {
        handler(change);
      } catch (error) {
        console.error(`调试处理器 ${name} 执行失败:`, error);
      }
    }
  }
  
  getEventHistory() {
    return this.capturedEvents;
  }
  
  analyzePerformance() {
    const events = this.capturedEvents;
    if (events.length < 2) return null;
    
    const analysis = {
      totalEvents: events.length,
      eventTypes: {},
      processingTimes: [],
      averageEventsPerSecond: 0
    };
    
    // 分析事件类型分布
    events.forEach(event => {
      analysis.eventTypes[event.operationType] = 
        (analysis.eventTypes[event.operationType] || 0) + 1;
    });
    
    // 计算处理时间
    for (let i = 1; i < events.length; i++) {
      const processingTime = events[i].timestamp - events[i-1].timestamp;
      analysis.processingTimes.push(processingTime);
    }
    
    // 计算平均每秒事件数
    const timeWindow = events[events.length-1].timestamp - events[0].timestamp;
    analysis.averageEventsPerSecond = events.length / (timeWindow / 1000);
    
    return analysis;
  }
  
  generateReport() {
    const performance = this.analyzePerformance();
    const report = {
      generatedAt: new Date().toISOString(),
      totalEventsCaptured: this.capturedEvents.length,
      performanceAnalysis: performance,
      recentErrors: this.capturedEvents.filter(e => e.error).slice(-10),
      eventTypeDistribution: performance ? performance.eventTypes : {}
    };
    
    return report;
  }
}

通过这个全面的指南，您应该能够深入理解 MongoDB Change Streams 的强大功能，并在实际生产环境中有效地使用它们来构建实时数据处理系统。记住，Change Streams 是 MongoDB 生态系统中非常强大的工具，正确使用可以极大地提升应用程序的实时能力和用户体验。