MongoDB数据库迁移工具实战对比记录企业迁移过程中遇到的同步失败问题和解决方案

一、前言：我们为什么要折腾这个

说出来你可能不信，去年我们团队经历了一场”噩梦级”的MongoDB迁移。从AWS的云服务迁回自建机房，涉及12个库、47张集合、总数据量接近15TB。说实话，刚开始大家都觉得”不就是复制粘贴嘛”，结果真正干起来才发现，水深得离谱。

同步失败、数据类型丢失、索引重建失败、时间戳偏差、主从延迟……每一个坑都能让你怀疑人生。这篇文章就是把我们的血泪史整理出来，希望能帮到正在或者即将面临同样困境的你们。

二、迁移工具横评：谁才是真命天子

2.1 主流工具一览

我们在迁移前调研了市面上主流的MongoDB迁移工具，大概列了这么一张表：

工具名称	类型	优势	劣势	适用场景
MongoDB官方mongodump/mongorestore	离线备份恢复	官方支持、稳定可靠	不支持增量、停服时间长	小规模迁移
MongoDB官方 mongosync	在线同步	官方出品、增量同步	配置复杂、文档少	中型规模
Percona MongoDB Toolkit	工具集	功能丰富、社区活跃	部分功能收费	生产环境
MongoChef (Studio 3T)	GUI工具	界面友好、操作简单	大数据量性能差	小数据量
Debezium + MongoDB Connector	CDC工具	实时同步、开源免费	需要Kafka配合、架构复杂	实时数据同步
CloudBeaver + Mongostat	监控+同步	可视化好	功能单一	辅助工具
自建Python脚本	自定义开发	灵活可控	开发成本高	特殊需求

2.2 我们的选型过程

我们最终采用的是 混合方案：核心大库用官方 mongodump + mongorestore，增量同步用自定义的 Change Streams 方案，小库用 MongoChef 辅助迁移。

为什么这么选？因为没有任何一个工具能完美解决所有场景。

三、实战记录：我们遇到的坑

3.1 坑一：`bson` 时间戳精度丢失

问题描述

迁移后发现，部分文档的 createdAt 字段时间戳出现了偏差，偏移量在毫秒级别。

原因分析

MongoDB的 Date 类型在存储时精度是毫秒，但在某些旧版客户端驱动中，序列化时会截断精度。我们的旧应用使用的是 mongo-java-driver 3.4.x 版本，在反序列化时出现了这个问题。

解决方案

# 错误做法：直接用JSON序列化和反序列化
import json
from datetime import datetime

# 这会丢失精度
data = {'name': 'test', 'createdAt': datetime.now()}
json_str = json.dumps(data, default=str)
restored = json.loads(json_str)

# 正确做法：使用bson库正确处理Date类型
from bson import datetime as bson_datetime
from bson import CodecOptions
from pymongo import MongoClient
import json

client = MongoClient('mongodb://localhost:27017/', 
                     uuidRepresentation='standard',
                     codec_options=CodecOptions(document_class=dict))

# 确保时间戳以ISODate格式存储
doc = {
    'name': 'test',
    'createdAt': bson_datetime.utcnow()
}
collection.insert_one(doc)

# 读取时验证精度
read_doc = collection.find_one({'name': 'test'})
print(read_doc['createdAt'])  # 正确显示完整时间戳

// 如果是Node.js环境，使用正确的处理
const { ObjectId, Code } = require('mongodb');
const client = new MongoClient('mongodb://localhost:27017', {
  useNewUrlParser: true,
  useUnifiedTopology: true
});

async function migrateData() {
  const db = client.db('source_db');
  const sourceCollection = db.collection('users');
  const targetCollection = db.collection('users');
  
  // 使用bulkWrite保持时间戳精度
  const docs = await sourceCollection.find({}).toArray();
  
  const bulkOps = docs.map(doc => ({
    updateOne: {
      filter: { _id: doc._id },
      update: { $set: doc },
      upsert: true
    }
  }));
  
  await targetCollection.bulkWrite(bulkOps);
  console.log(`成功迁移 ${docs.length} 条记录`);
}

3.2 坑二：`ObjectId` 转换失败

问题描述

某些集合中存在字符串格式的 _id，迁移到新库后，查询语句因为类型不匹配而失败。

原因分析

MongoDB 允许 _id 使用不同类型，但最佳实践是使用 ObjectId。我们在迁移前的数据清洗阶段没有做充分的类型检查。

解决方案

// 检查并转换ObjectId
async function validateAndConvertObjectIds(dbName, collectionName) {
  const db = client.db(dbName);
  const collection = db.collection(collectionName);
  
  // 查找非ObjectId类型的记录
  const nonObjectIdDocs = await collection.find({
    _id: { $type: 'string' }  // 或者 $type: 'number'
  }).toArray();
  
  console.log(`发现 ${nonObjectIdDocs.length} 条非ObjectId记录`);
  
  // 转换字符串为ObjectId
  const bulkOps = nonObjectIdDocs.map(doc => ({
    updateOne: {
      filter: { _id: doc._id },
      update: {
        $set: { 
          _id: new ObjectId(doc._id),
          _idConvertedAt: new Date()
        },
        $unset: { tempId: "" }  // 如果有的话
      }
    }
  }));
  
  if (bulkOps.length > 0) {
    const result = await collection.bulkWrite(bulkOps);
    console.log(`成功转换 ${result.modifiedCount} 条记录`);
  }
}

// 迁移前调用
await validateAndConvertObjectIds('source_db', 'users');
await validateAndConvertObjectIds('source_db', 'orders');

# Python版本的处理
from bson import ObjectId
from pymongo import MongoClient
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def convert_string_objectids(db_name, collection_name):
    """将字符串类型的_id转换为ObjectId"""
    client = MongoClient('mongodb://localhost:27017/')
    db = client[db_name]
    collection = db[collection_name]
    
    # 查找字符串类型的_id
    query = {'_id': {'$type': 'string'}}
    docs = collection.find(query)
    
    bulk_operations = []
    for doc in docs:
        try:
            # 验证是否是有效的ObjectId
            if ObjectId.is_valid(doc['_id']):
                bulk_operations.append({
                    'update_one': {
                        'filter': {'_id': doc['_id']},
                        'update': {'$set': {'_id': ObjectId(doc['_id'])}}
                    }
                })
        except Exception as e:
            logger.error(f"转换失败: {doc['_id']}, 错误: {e}")
    
    if bulk_operations:
        result = collection.bulk_write(
            [op['update_one'] for op in bulk_operations]
        )
        logger.info(f"成功转换 {result.modified_count} 条记录")
    
    client.close()

3.3 坑三：集合命名空间过长

问题描述

旧库中有些集合名超过了MongoDB的限制长度（128字节），迁移到新版本的MongoDB 5.0+后直接报错。

原因分析

MongoDB 5.0开始，集合名的最大长度从之前的限制变为了更严格的128字节（UTF-8编码）。我们的旧集合中有使用中文命名的情况，比如 "用户订单历史表"，在UTF-8下超出了限制。

解决方案

// 检查集合名长度
async function checkCollectionNames(dbName) {
  const db = client.db(dbName);
  const collections = await db.listCollections().toArray();
  
  const oversized = collections.filter(col => {
    // MongoDB使用UTF-8编码，每个中文字符占3字节
    const buffer = Buffer.from(col.name, 'utf8');
    return buffer.length > 128;
  });
  
  if (oversized.length > 0) {
    console.warn('发现超长的集合名：');
    oversized.forEach(col => {
      const buffer = Buffer.from(col.name, 'utf8');
      console.log(`  ${col.name} (${buffer.length} bytes)`);
    });
  }
  
  return oversized;
}

// 重命名策略
async function renameOversizedCollections(dbName, collectionName, newName) {
  const db = client.db(dbName);
  const originalBuffer = Buffer.from(collectionName, 'utf8');
  const newBuffer = Buffer.from(newName, 'utf8');
  
  if (newBuffer.length > 128) {
    throw new Error(`新名称 "${newName}" 超出长度限制 (${newBuffer.length} bytes)`);
  }
  
  // 执行重命名
  await db.renameCollection(collectionName, newName);
  console.log(`成功重命名: ${collectionName} -> ${newName}`);
}

# Python版本检查
import re

def check_collection_names(db_name):
    """检查MongoDB集合名长度"""
    client = MongoClient('mongodb://localhost:27017/')
    db = client[db_name]
    
    oversized = []
    for col in db.list_collections():
        name = col['name']
        name_bytes = name.encode('utf-8')
        
        if len(name_bytes) > 128:
            oversized.append({
                'name': name,
                'bytes': len(name_bytes),
                'reason': '超过128字节限制'
            })
            print(f"⚠️ 超长集合名: {name} ({len(name_bytes)} bytes)")
    
    client.close()
    return oversized

3.4 坑四：索引重建失败

问题描述

迁移后，部分集合的索引重建失败，查询性能大幅下降。

原因分析

有两个主要原因：

索引键类型不一致：某些文档的索引字段类型不同
索引选项冲突：旧库使用了特殊的索引选项，新库不支持

解决方案

// 迁移前检查和重建索引
async function analyzeAndRebuildIndexes(dbName, collectionName) {
  const db = client.db(dbName);
  const collection = db.collection(collectionName);
  
  // 获取现有索引信息
  const indexes = await collection.indexes();
  console.log(`集合 ${collectionName} 现有索引:`);
  indexes.forEach(idx => {
    console.log(`  - ${JSON.stringify(idx.key)} ${idx.options ? JSON.stringify(idx.options) : ''}`);
  });
  
  // 检查索引键的类型一致性
  const sampleDocs = await collection.find({}).limit(1000).toArray();
  const typeMap = {};
  
  sampleDocs.forEach(doc => {
    indexes.forEach(idx => {
      Object.keys(idx.key).forEach(field => {
        if (doc[field] !== undefined) {
          const fieldType = typeof doc[field];
          if (!typeMap[field]) {
            typeMap[field] = new Set();
          }
          // 特殊处理：ObjectId和string可能看起来不同但需要兼容
          if (fieldType === 'object' && doc[field] && doc[field].constructor && 
              doc[field].constructor.name === 'ObjectId') {
            typeMap[field].add('ObjectId');
          } else {
            typeMap[field].add(fieldType);
          }
        }
      });
    });
  });
  
  // 输出类型不一致的字段
  Object.keys(typeMap).forEach(field => {
    const types = Array.from(typeMap[field]);
    if (types.length > 1) {
      console.warn(`⚠️ 字段 ${field} 存在多种类型: ${types.join(', ')}`);
    }
  });
  
  // 重建索引（在新库上）
  for (const idx of indexes) {
    try {
      // 跳过系统索引
      if (idx.name === '_id_') continue;
      
      await collection.createIndex(idx.key, {
        ...idx.options,
        background: true  // 后台创建，不阻塞写入
      });
      console.log(`✅ 索引创建成功: ${JSON.stringify(idx.key)}`);
    } catch (err) {
      console.error(`❌ 索引创建失败: ${JSON.stringify(idx.key)}, 错误: ${err.message}`);
      // 记录失败的索引，稍后手动处理
    }
  }
}

// 针对特定问题的索引重建
async function fixProblematicIndexes(dbName, collectionName) {
  const db = client.db(dbName);
  const collection = db.collection(collectionName);
  
  // 1. 删除有问题的索引
  const badIndexes = await collection.indexes();
  for (const idx of badIndexes) {
    if (idx.key.some((val, key) => {
      // 检查是否有非法的索引键
      return typeof val === 'object' && val !== null;
    })) {
      await collection.dropIndex(idx.name);
      console.log(`删除问题索引: ${idx.name}`);
    }
  }
  
  // 2. 重建文本索引
  await collection.createIndex(
    { description: 'text', tags: 'text' },
    { 
      weights: { description: 10, tags: 5 },
      default_language: 'chinese',  // 中文分词
      background: true
    }
  );
}

# Python版本
def analyze_indexes(db_name, collection_name):
    """分析集合索引"""
    client = MongoClient('mongodb://localhost:27017/')
    db = client[db_name]
    collection = db[collection_name]
    
    # 获取索引列表
    indexes = list(collection.index_information().values())
    
    print(f"\n=== 集合 {collection_name} 索引分析 ===")
    for idx in indexes:
        key = idx.get('key', [])
        name = idx.get('name', 'unknown')
        print(f"索引: {name}")
        print(f"  键: {key}")
        print(f"  唯一: {idx.get('unique', False)}")
        print(f"  过期: {idx.get('expireAfterSeconds', '永不过期')}秒")
    
    client.close()

3.5 坑五：Change Streams 断连重连

问题描述

在增量同步过程中，Change Streams 频繁断连，导致数据同步中断。

原因分析

网络不稳定：跨机房同步，网络延迟和抖动
会话超时：默认会话超时时间过短
游标丢失：MongoDB 的 Change Streams 游标有最大存活时间

解决方案

// 健壮的Change Streams实现
class RobustChangeStream {
  constructor(mongodbUri, dbName, collectionNames) {
    this.uri = mongodbUri;
    this.dbName = dbName;
    this.collectionNames = Array.isArray(collectionNames) 
      ? collectionNames 
      : [collectionNames];
    this.client = null;
    this.changeStream = null;
    this.resumeToken = null;
    this.maxRetries = 5;
    this.retryDelay = 3000; // 3秒
  }

  async connect() {
    const { MongoClient } = require('mongodb');
    this.client = new MongoClient(this.uri, {
      maxPoolSize: 10,
      serverSelectionTimeoutMS: 5000,
      socketTimeoutMS: 45000,
      heartbeatFrequencyMS: 10000,
    });
    await this.client.connect();
    console.log('✅ 连接到MongoDB成功');
  }

  async startWithResume() {
    const db = this.client.db(this.dbName);
    
    // 使用resumeAfter如果之前有记录
    const pipeline = [
      {
        $match: {
          operationType: { $in: ['insert', 'update', 'replace', 'delete'] }
        }
      },
      {
        $project: {
          documentKey: 1,
          operationType: 1,
          fullDocument: 1,
          updateDescription: 1
        }
      }
    ];

    const options = {
      fullDocument: 'updateLookup',  // 获取更新后的完整文档
      maxAwaitTimeMS: 1000,          // 最大等待时间
      collateralAllowlistCheck: false
    };

    if (this.resumeToken) {
      options.resumeAfter = this.resumeToken;
    }

    this.changeStream = db.watch(pipeline, options);
    
    this.changeStream.on('change', (change) => {
      this.resumeToken = change._id;  // 保存resume token
      this.processChange(change);
    });

    this.changeStream.on('error', async (error) => {
      console.error('❌ Change Stream错误:', error.message);
      await this.handleStreamError(error);
    });

    this.changeStream.on('close', () => {
      console.log('🔌 Change Stream关闭，尝试重新连接...');
      this.handleStreamClose();
    });

    console.log('✅ Change Stream开始监听');
  }

  async processChange(change) {
    try {
      const { operationType, documentKey, fullDocument, updateDescription } = change;
      
      switch (operationType) {
        case 'insert':
          await this.handleInsert(documentKey, fullDocument);
          break;
        case 'update':
          await this.handleUpdate(documentKey, updateDescription, fullDocument);
          break;
        case 'replace':
          await this.handleReplace(documentKey, fullDocument);
          break;
        case 'delete':
          await this.handleDelete(documentKey);
          break;
        default:
          console.log(`未知操作类型: ${operationType}`);
      }
    } catch (error) {
      console.error('处理变更事件失败:', error);
      // 不要抛出错误，避免关闭Change Stream
    }
  }

  async handleStreamError(error) {
    // 检查是否是resume token失效
    if (error.code === 40576 || error.code === 40577) {
      console.warn('Resume token已过期，尝试从最近的事件恢复...');
      this.resumeToken = null;
      
      // 等待一段时间后重试
      await this.sleep(this.retryDelay);
      await this.startWithResume();
      return;
    }
    
    // 其他错误，尝试重连
    await this.sleep(this.retryDelay);
    await this.startWithResume();
  }

  handleStreamClose() {
    // 延迟重连
    setTimeout(() => {
      this.startWithResume().catch(err => {
        console.error('重连失败:', err);
      });
    }, this.retryDelay);
  }

  handleInsert(documentKey, fullDocument) {
    // 发送到目标库
    console.log(`插入: ${documentKey._id}`);
    // 异步处理，不阻塞
    this.sendToTarget('insert', documentKey, fullDocument);
  }

  handleUpdate(documentKey, updateDescription, fullDocument) {
    console.log(`更新: ${documentKey._id}`);
    this.sendToTarget('update', documentKey, fullDocument);
  }

  handleReplace(documentKey, fullDocument) {
    console.log(`替换: ${documentKey._id}`);
    this.sendToTarget('replace', documentKey, fullDocument);
  }

  handleDelete(documentKey) {
    console.log(`删除: ${documentKey._id}`);
    this.sendToTarget('delete', documentKey, null);
  }

  sendToTarget(operation, documentKey, fullDocument) {
    // 实现具体的同步逻辑
    // 可以异步发送到Redis、Kafka或直接写入目标库
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  async close() {
    if (this.changeStream) {
      this.changeStream.close();
    }
    if (this.client) {
      await this.client.close();
    }
  }
}

// 使用示例
async function main() {
  const changeStream = new RobustChangeStream(
    'mongodb://localhost:27017',
    'myDatabase',
    ['users', 'orders', 'products']
  );
  
  await changeStream.connect();
  await changeStream.startWithResume();
  
  // 优雅退出
  process.on('SIGINT', async () => {
    console.log('\n正在关闭Change Stream...');
    await changeStream.close();
    process.exit(0);
  });
}

main().catch(console.error);

3.6 坑六：大字段（超过16MB）处理

问题描述

部分文档包含较大的文本或图片数据，超过了MongoDB单文档16MB的限制。

原因分析

MongoDB 的单文档大小限制是硬性限制，超过16MB的文档需要特殊处理。

解决方案

// 检测大字段文档
async function detectLargeDocuments(dbName, collectionName) {
  const db = client.db(dbName);
  const collection = db.collection(collectionName);
  
  // 使用countDocuments统计，避免加载全部数据
  const totalDocs = await collection.countDocuments();
  
  // 抽样检查
  const sampleSize = Math.min(1000, totalDocs);
  const sampleDocs = await collection.find({})
    .project({ _id: 1, data: 1, content: 1, description: 1 })
    .limit(sampleSize)
    .toArray();
  
  const largeDocs = [];
  
  for (const doc of sampleDocs) {
    const docString = JSON.stringify(doc);
    const sizeInBytes = Buffer.byteLength(docString, 'utf8');
    
    if (sizeInBytes > 10 * 1024 * 1024) {  // 大于10MB
      largeDocs.push({
        _id: doc._id,
        sizeMB: (sizeInBytes / 1024 / 1024).toFixed(2),
        estimatedField: detectLargeField(doc)
      });
    }
  }
  
  if (largeDocs.length > 0) {
    console.warn(`发现 ${largeDocs.length} 个大字段文档:`);
    largeDocs.forEach(doc => {
      console.log(`  ID: ${doc._id}, 大小: ${doc.sizeMB}MB, 疑似字段: ${doc.estimatedField}`);
    });
  }
  
  return largeDocs;
}

function detectLargeField(doc) {
  const fields = Object.keys(doc);
  let maxSize = 0;
  let maxField = '';
  
  for (const field of fields) {
    if (field === '_id') continue;
    const value = doc[field];
    const size = Buffer.byteLength(JSON.stringify(value), 'utf8');
    if (size > maxSize) {
      maxSize = size;
      maxField = field;
    }
  }
  
  return maxField;
}

// 分片存储方案
async function splitLargeDocument(dbName, collectionName, documentId, targetDb) {
  const sourceDb = client.db(dbName);
  const targetDbClient = new MongoClient('mongodb://localhost:27018/');
  await targetDbClient.connect();
  const targetDbInstance = targetDbClient.db(targetDb);
  
  const collection = sourceDb.collection(collectionName);
  const doc = await collection.findOne({ _id: documentId });
  
  if (!doc) {
    throw new Error(`文档 ${documentId} 不存在`);
  }
  
  // 分离大字段到单独集合
  const largeField = detectLargeField(doc);
  const mainDoc = { ...doc };
  delete mainDoc[largeField];
  
  // 存储主文档
  const mainCollection = targetDbInstance.collection(`${collectionName}_main`);
  await mainCollection.insertOne(mainDoc);
  
  // 存储大字段
  const blobCollection = targetDbInstance.collection(`${collectionName}_blobs`);
  await blobCollection.insertOne({
    _id: documentId,
    fieldName: largeField,
    data: doc[largeField],
    dataSize: Buffer.byteLength(JSON.stringify(doc[largeField]), 'utf8'),
    createdAt: new Date()
  });
  
  console.log(`✅ 文档 ${documentId} 已分片存储`);
  
  await targetDbClient.close();
}

3.7 坑七：副本集选举期间同步中断

问题描述

在迁移过程中，源库发生主节点选举，导致同步中断。

原因分析

MongoDB 副本集在选举期间会短暂不可写（约10-30秒），如果在这期间进行同步操作，会失败。

解决方案

// 选举期间自动重试的同步器
class ElectionResilientSync {
  constructor(sourceUri, targetUri, dbName, collectionName) {
    this.sourceClient = new MongoClient(sourceUri, {
      readPreference: 'secondaryPreferred'  // 优先读从节点
    });
    this.targetClient = new MongoClient(targetUri);
    this.dbName = dbName;
    this.collectionName = collectionName;
    this.isSyncing = false;
  }

  async connect() {
    await this.sourceClient.connect();
    await this.targetClient.connect();
  }

  async syncWithRetry(doc, maxRetries = 5) {
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        await this.performSync(doc);
        console.log(`✅ 同步成功 (尝试 ${attempt}/${maxRetries})`);
        return true;
      } catch (error) {
        if (attempt === maxRetries) {
          console.error(`❌ 同步最终失败: ${error.message}`);
          return false;
        }
        
        // 检查是否是选举相关的错误
        if (this.isElectionError(error)) {
          console.warn(`⚠️ 检测到副本集选举，等待后重试 (${attempt}/${maxRetries})`);
          await this.sleep(2000 * attempt);  // 指数退避
        } else {
          throw error;  // 非选举错误直接抛出
        }
      }
    }
  }

  isElectionError(error) {
    const electionErrors = [
      10107,  // NotYetInitialized
      10058,  // NotPrimary
      11602,  // NotPrimaryNoSecondaryOk
      18940,  // NodeAtPrimaryState
      13435,  // NodeIsSplitBrainCandidate
    ];
    return electionErrors.includes(error.code);
  }

  async performSync(doc) {
    const sourceDb = this.sourceClient.db(this.dbName);
    const targetDb = this.targetClient.db(this.dbName);
    
    const sourceCollection = sourceDb.collection(this.collectionName);
    const targetCollection = targetDb.collection(this.collectionName);
    
    // 使用upsert确保幂等性
    await targetCollection.updateOne(
      { _id: doc._id },
      { $set: doc },
      { upsert: true }
    );
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  async close() {
    await this.sourceClient.close();
    await this.targetClient.close();
  }
}

// 使用示例
async function main() {
  const sync = new ElectionResilientSync(
    'mongodb://primary:27017,secondary1:27017,secondary2:27017',
    'mongodb://target:27017',
    'production_db',
    'users'
  );
  
  await sync.connect();
  
  // 从Change Stream读取并同步
  const db = sync.sourceClient.db('production_db');
  const changeStream = db.collection('users').watch();
  
  for await (const change of changeStream) {
    await sync.syncWithRetry(change.fullDocument);
  }
  
  await sync.close();
}

四、迁移前的 checklist

在我们踩了这么多坑之后，总结了一份迁移前的 checklist，建议大家在迁移前逐项确认：

4.1 数据层面检查

□ 所有集合的文档数量统计完成
□ 发现并处理超过16MB的文档
□ 所有ObjectId字段验证并转换
□ 所有Date类型字段检查精度
□ 检查并修复索引冲突
□ 确认副本集状态和选举机制
□ 备份源库完整数据
□ 验证源库和目标库的MongoDB版本兼容性

4.2 网络层面检查

□ 确认源库和目标库网络连通性
□ 测试延迟和带宽是否满足需求
□ 确认防火墙规则开放相应端口
□ 准备网络中断时的应急预案
□ 测试DNS解析是否正常

4.3 工具层面检查

□ 选择适合的迁移工具组合
□ 准备迁移脚本并测试
□ 准备回滚方案
□ 确认监控和日志系统就绪
□ 准备数据校验脚本

五、数据校验：如何确认迁移成功

迁移完成不代表结束，数据校验才是关键。我们开发了一套校验工具：

// 数据校验工具
class MigrationValidator {
  constructor(sourceClient, targetClient) {
    this.sourceClient = sourceClient;
    this.targetClient = targetClient;
  }

  // 统计校验：对比文档数量
  async validateCounts(dbName, collectionName) {
    const sourceDb = this.sourceClient.db(dbName);
    const targetDb = this.targetClient.db(dbName);
    
    const sourceCount = await sourceDb.collection(collectionName).countDocuments();
    const targetCount = await targetDb.collection(collectionName).countDocuments();
    
    const isMatch = sourceCount === targetCount;
    console.log(`[${collectionName}] 源库: ${sourceCount}, 目标库: ${targetCount}, ${isMatch ? '✅ 一致' : '❌ 不一致'}`);
    
    return isMatch;
  }

  // 抽样校验：随机抽取文档比对
  async validateSampling(dbName, collectionName, sampleSize = 100) {
    const sourceDb = this.sourceClient.db(dbName);
    const targetDb = this.targetClient.db(dbName);
    
    const sourceCollection = sourceDb.collection(collectionName);
    const targetCollection = targetDb.collection(collectionName);
    
    // 获取源库随机文档
    const sourceDocs = await sourceCollection.aggregate([
      { $sample: { size: sampleSize } },
      { $project: { _id: 1 } }
    ]).toArray();
    
    let mismatches = 0;
    
    for (const doc of sourceDocs) {
      const sourceDoc = await sourceCollection.findOne({ _id: doc._id });
      const targetDoc = await targetCollection.findOne({ _id: doc._id });
      
      if (!targetDoc) {
        console.log(`❌ 目标库缺少文档: ${doc._id}`);
        mismatches++;
        continue;
      }
      
      // 深度比较（忽略_metadata等字段）
      const sourceClean = this.cleanForComparison(sourceDoc);
      const targetClean = this.cleanForComparison(targetDoc);
      
      if (JSON.stringify(sourceClean) !== JSON.stringify(targetClean)) {
        console.log(`❌ 文档内容不一致: ${doc._id}`);
        mismatches++;
      }
    }
    
    const passRate = ((sampleSize - mismatches) / sampleSize * 100).toFixed(2);
    console.log(`[${collectionName}] 抽样校验: ${passRate}% 一致 (${mismatches} 个差异)`);
    
    return mismatches === 0;
  }

  cleanForComparison(doc) {
    const clean = { ...doc };
    delete clean._meta;
    delete clean._version;
    delete clean.migratedAt;
    return clean;
  }

  // 校验摘要
  async generateValidationReport(dbName, collections) {
    const report = {
      database: dbName,
      timestamp: new Date().toISOString(),
      collections: {}
    };
    
    for (const colName of collections) {
      report.collections[colName] = {
        countMatch: await this.validateCounts(dbName, colName),
        sampleValid: await this.validateSampling(dbName, colName)
      };
    }
    
    const allPassed = Object.values(report.collections).every(
      col => col.countMatch && col.sampleValid
    );
    
    report.overallResult = allPassed ? '✅ 全部通过' : '❌ 存在差异';
    
    return report;
  }
}

// 使用示例
async function runValidation() {
  const sourceClient = new MongoClient('mongodb://source:27017');
  const targetClient = new MongoClient('mongodb://target:27017');
  
  await sourceClient.connect();
  await targetClient.connect();
  
  const validator = new MigrationValidator(sourceClient, targetClient);
  
  const report = await validator.generateValidationReport('mydb', [
    'users', 'orders', 'products', 'transactions'
  ]);
  
  console.log('\n=== 迁移校验报告 ===');
  console.log(JSON.stringify(report, null, 2));
  
  await sourceClient.close();
  await targetClient.close();
}

六、经验总结与反思

说实话，这次迁移让我们团队深刻认识到几个要点：

没有银弹工具：没有任何一个工具能完美解决所有迁移场景，混合方案才是王道
预处理是关键：迁移前花一周做数据清洗，比迁移后花一个月修bug划算得多
监控和日志不能省：每一步操作都要有日志，出了问题才能快速定位
回滚方案要有：任何迁移都可能失败，没有回滚方案就是赌博
测试环境要充分模拟：生产环境的复杂性远超测试环境，尽量在测试环境模拟真实场景

最后想说，数据迁移是一场硬仗，但只要准备充分、步步为营，总能拿下。希望这篇记录能给正在或者即将踏上迁移之路的你们一些参考。如果你们遇到了我们没提到的问题，欢迎一起交流讨论。

正文

MongoDB数据库迁移工具实战对比记录企业迁移过程中遇到的同步失败问题和解决方案

MongoDB数据库迁移工具实战对比记录企业迁移过程中遇到的同步失败问题和解决方案

一、前言：我们为什么要折腾这个

二、迁移工具横评：谁才是真命天子

2.1 主流工具一览

2.2 我们的选型过程

三、实战记录：我们遇到的坑

3.1 坑一：`bson` 时间戳精度丢失

3.2 坑二：`ObjectId` 转换失败

3.3 坑三：集合命名空间过长

3.4 坑四：索引重建失败

3.5 坑五：Change Streams 断连重连

3.6 坑六：大字段（超过16MB）处理

3.7 坑七：副本集选举期间同步中断

四、迁移前的 checklist

4.1 数据层面检查

4.2 网络层面检查

4.3 工具层面检查

五、数据校验：如何确认迁移成功

六、经验总结与反思

相关阅读

如何轻松实现MongoDB数据迁移，解决常见迁移难题与技巧详解

Fedora系统迁移：轻松上手，完整步骤解析，告别繁琐，体验流畅过渡

轻松实现SQLite数据库升级：盘点实用迁移工具与操作指南

如何轻松实现MongoDB数据迁移，避免数据丢失与中断？实战指南助你一臂之力

Fedora系统迁移：轻松上手，一键迁移，告别繁琐操作，体验流畅新系统

轻松迁移MongoDB：五大实用工具，轻松实现数据无缝转移与同步

如何轻松实现SQL数据迁移：五大实用工具推荐与实操步骤详解

从SQLite新手到高级玩家：轻松掌握数据库迁移工具全攻略

掌握MongoDB迁移：盘点实用工具与操作指南

轻松掌握SQL数据迁移：四大热门工具全面评测与实操指南

MongoDB数据库迁移工具实战对比记录企业迁移过程中遇到的同步失败问题和解决方案

一、前言：我们为什么要折腾这个

二、迁移工具横评：谁才是真命天子

2.1 主流工具一览

2.2 我们的选型过程

三、实战记录：我们遇到的坑

3.1 坑一：bson 时间戳精度丢失

3.2 坑二：ObjectId 转换失败

3.3 坑三：集合命名空间过长

3.4 坑四：索引重建失败

3.5 坑五：Change Streams 断连重连

3.6 坑六：大字段（超过16MB）处理

3.7 坑七：副本集选举期间同步中断

四、迁移前的 checklist

4.1 数据层面检查

4.2 网络层面检查

4.3 工具层面检查

五、数据校验：如何确认迁移成功

六、经验总结与反思

相关阅读

如何轻松实现MongoDB数据迁移，解决常见迁移难题与技巧详解

Fedora系统迁移：轻松上手，完整步骤解析，告别繁琐，体验流畅过渡

轻松实现SQLite数据库升级：盘点实用迁移工具与操作指南

如何轻松实现MongoDB数据迁移，避免数据丢失与中断？实战指南助你一臂之力

Fedora系统迁移：轻松上手，一键迁移，告别繁琐操作，体验流畅新系统

轻松迁移MongoDB：五大实用工具，轻松实现数据无缝转移与同步

如何轻松实现SQL数据迁移：五大实用工具推荐与实操步骤详解

从SQLite新手到高级玩家：轻松掌握数据库迁移工具全攻略

掌握MongoDB迁移：盘点实用工具与操作指南

轻松掌握SQL数据迁移：四大热门工具全面评测与实操指南

3.1 坑一：`bson` 时间戳精度丢失

3.2 坑二：`ObjectId` 转换失败