为什么我要跟你聊聊这件事
说实话，两年前我在一家做内容电商的公司，当时我们的技术栈还是传统的 MySQL 5.7 堆栈。随着用户量从几十万涨到几千万，我们遇到了一个典型的“成长烦恼”：订单表、商品表、用户行为日志这些数据结构越来越松散，有的字段是 JSON，有的字段是变长的字符串，MySQL 的 Schema 约束让我们每次加字段都得小心翼翼，甚至还要申请停机窗口做 DDL 变更。
那个时间点，CTO 拍板决定：“上 MongoDB，试试 NoSQL。”
听起来很美好对吧？但现实给了我们狠狠一击。迁移过程中，数据一致性丢了、查询性能反了、甚至有一次因为类型转换错误，导致线上订单金额显示为 null，客诉量直接飙升至平时的五十倍。
这篇指南不是那种“Hello World”式的教程，而是我带着团队踩过所有坑之后，总结出来的实战血泪史。无论你是正在规划迁移的架构师，还是准备跳槽面试的前端/后端开发，或者单纯想了解 NoSQL 的朋友，这篇文章都能帮你把这件事彻底讲清楚。我们从头到尾，一步步拆解，保证你看完之后，对“零停机迁移”有个清晰、可落地的认知。
第一阶段：思维转变——别把 MongoDB 当 MySQL 用
很多人迁移失败，根源不在技术，而在思维。
如果你带着 MySQL 的习惯去写 MongoDB，那你大概率会出事。我给你举几个最常见的错误认知：
1. “我没有外键，所以我不需要关心数据一致性”
错。MySQL 有外键约束，数据库层面帮你保证了 user_id 在用户表里一定存在。MongoDB 没有外键，这意味着什么？意味着如果你在订单文档里写了个 user_id: 12345，但你根本不去检查这个用户是否存在，你的业务逻辑就会出现幽灵订单。
正确姿势：在应用层做冗余校验，或者使用 MongoDB 的 $lookup（类似 JOIN）来关联数据，但要注意，$lookup 性能开销不小，慎用。
2. “聚合框架 (Aggregation Pipeline) 什么都能干”
MySQL 有 JOIN，有 GROUP BY，有子查询，很强大。MongoDB 也有聚合管道，但它的设计哲学是流式处理，每一步都要控制内存使用。如果你试图用聚合管道实现一个复杂的、多表关联的、带大量排序的业务查询，你会发现它比 MySQL 慢几十倍，甚至直接报错 Exceeded memory limit。
正确姿势：先问自己，这个查询在 MongoDB 里真的合适吗？如果需要复杂的多表关联分析，考虑把数据同步回 MySQL 做 OLAP，或者用专门的 BI 工具。MongoDB 擅长的是单文档内的复杂嵌套查询和高并发写入。
3. “索引越多越好”
MySQL 里，索引确实能加速查询，但也会拖慢写入。MongoDB 也一样，但更严重。MongoDB 是列存逻辑、行存物理的变种，每个索引都会占用额外空间，并且在写入时维护索引树。如果你在一个文本文字段上加了索引，你的写性能会跌到谷底。
正确姿势：只索引你真正需要查询的字段，而且是等值查询或范围查询。别为了好看加索引。
第二阶段：迁移前的“体检”——评估与准备
别急着动手写代码，先坐下来，把数据摸清楚。
2.1 数据模型映射
这是最关键的一步。MySQL 是关系型，MongoDB 是文档型。你需要决定：是把一个 MySQL 表映射成一个 MongoDB 集合，还是把多个表合并成一个文档？
案例：用户表 vs 用户地址表
在 MySQL 里，你通常会有两张表：users 和 addresses，通过 user_id 关联。
错误做法：1:1 映射，每个地址一个文档，查询时应用层多次查询。
推荐做法：嵌入文档。把用户的常用地址（比如最多3个）直接嵌入到用户文档中：
{
"_id": ObjectId("507f1f77bcf86cd7994f1c0d"),
"username": "zhangsan",
"email": "zhangsan@example.com",
"addresses": [
{
"type": "home",
"province": "Beijing",
"city": "Beijing",
"detail": "Chaoyang District...",
"isDefault": true
},
{
"type": "office",
"province": "Beijing",
"city": "Beijing",
"detail": "Haidian District...",
"isDefault": false
}
]
}
这样做的好处是一次查询拿到所有数据，减少网络 IO。但注意，MongoDB 单个文档最大限制是 16MB，如果地址数量无限增长，你就要改成引用，用 user_id 关联查询。
决策树：
- 数据是否经常一起读写？ -> 是，嵌入。
- 数据量是否巨大且不确定上限？ -> 否，引用。
- 数据是否会被多个实体共享？ -> 是，引用。
2.2 类型转换陷阱
MySQL 有 INT、VARCHAR、DATETIME，MongoDB 有 NumberInt、String、Date。但问题在于，MySQL 有时很宽容。
比如，MySQL 里存了一个日期字符串 "2023-10-01"，你可以直接做日期比较。但在 MongoDB 里，如果你把这个字段存为字符串，然后尝试 $gt 操作，它是按字典序比较的，结果会完全错误。
必须做的检查：
- 扫描所有表中，日期字段是否混存了字符串和日期类型？
- 数值字段是否混存了整数和浮点数？（比如
1和1.0在 JSON 里是不同的，但在 JavaScript 里可能一样） - 布尔字段是否用了字符串
"true"和"false"？
建议：在迁移脚本里，加一个严格的数据清洗层，把所有日期统一转为 ISODate 对象，所有数值统一类型。
第三阶段：零停机迁移的核心——双写与数据同步
这是整个迁移中最难、也是最关键的部分。零停机意味着在迁移过程中，线上业务不能停，数据不能丢。
我们采用的方案是：CDC（Change Data Capture）+ 双写 + 数据比对 + 灰度切流。
3.1 CDC 技术选型
CDC 就是捕获 MySQL 的数据变更，然后同步到 MongoDB。
主流方案有两个：
- Canal（阿里开源，基于 MySQL binlog 解析）
- Debezium（开源，支持多种数据库，基于 binlog/ WAL）
我们选择了 Canal，因为它对 MySQL 支持最好，生态成熟，社区活跃。
架构示意：
MySQL -> Binlog -> Canal Server -> Kafka -> Canal Client (写 MongoDB)
3.2 实施步骤
步骤一：搭建 Canal 同步链路
首先，你需要修改 MySQL 配置，开启 binlog：
# my.cnf
[mysqld]
log-bin=mysql-bin
binlog-format=ROW # 必须是 ROW 模式，才能捕获到行级变更
server-id=1
重启 MySQL 后，启动 Canal Server，并配置订阅你的数据库和表。
步骤二：开发数据同步程序
你需要写一个程序，监听 Canal 消息，然后将其转换为 MongoDB 的写入操作。
这里有一个关键问题：如何处理 MySQL 的 UPDATE 语句？
在 MySQL 里，UPDATE users SET status=1 WHERE id=100，这条语句在 binlog 里可能只记录了变更的字段。但在 MongoDB 里，如果你只更新 status 字段，其他的字段会不会被覆盖？
答案：这取决于你的更新策略。
策略 A：全量覆盖（推荐用于初始同步） 每次 MySQL 变更，都把整个文档拉取出来，修改后整体写入 MongoDB。这样最安全，不会丢失字段。
策略 B：部分更新 只更新变更的字段。这性能好，但风险高，因为 MongoDB 文档结构可能和 MySQL 表结构不完全一致。
我们的做法：混合模式。
- 首次全量迁移：通过脚本，把 MySQL 历史数据全部导入 MongoDB，采用全量覆盖。
- 增量同步：监听 binlog，对于
INSERT和DELETE，直接映射到 MongoDB 的insertOne和deleteOne。对于UPDATE，采用全量拉取 + 覆盖的方式，确保一致性。
# 伪代码示例：Canal 消息监听与 MongoDB 写入
import json
from canal.client import Client
from pymongo import MongoClient
client = Client()
client.connect(host='127.0.0.1', port=11111)
client.subscribe('test_db', 'users')
mongo_client = MongoClient('mongodb://localhost:27017/')
db = mongo_client['production_db']
collection = db['users']
while True:
messages = client.get(100)
for message in messages:
entry = message['entries']
for e in entry:
if e['entry_type'] == 'ROWDATA':
header = e['header']
event_type = e['entry_type']
if header.get('event_type') == 'INSERT':
# 解析 RowData，构建 MongoDB 文档
doc = parse_insert(e)
collection.insert_one(doc)
elif header.get('event_type') == 'UPDATE':
# 关键点：对于 UPDATE，先查最新数据，再覆盖
old_doc = parse_row(e['before'])
new_doc = parse_row(e['after'])
# 策略：全量覆盖，确保字段一致性
# 注意：_id 不能变
new_doc['_id'] = old_doc['_id']
collection.replace_one({'_id': old_doc['_id']}, new_doc)
elif header.get('event_type') == 'DELETE':
# 删除操作
doc = parse_row(e['before'])
collection.delete_one({'_id': doc['_id']})
步骤三：双写阶段
在 CDC 跑起来之后，我们不会立刻切换流量。我们会进入双写阶段。
什么是双写？ 应用层代码修改为：写 MySQL 的同时，也写 MongoDB。
// 伪代码：双写逻辑
public void saveUser(User user) {
// 1. 写 MySQL（主数据库，业务继续从这里读）
mysqlRepository.save(user);
// 2. 写 MongoDB（备用，用于后续查询或灰度）
mongoRepository.save(user);
// 3. 记录同步日志，用于后续比对
syncLogRepository.save(new SyncLog(user.getId(), "USER", "INSERT"));
}
为什么叫双写？ 因为此时，MySQL 和 MongoDB 里都有数据。但我们只从 MySQL 读，保证业务不受影响。
步骤四：数据比对
双写运行一段时间后，我们需要验证 MongoDB 的数据是否正确。
这里有一个笨但有效的方法：抽样比对。
每隔一段时间，随机抽取 1000 条 MySQL 记录，和 MongoDB 里对应的记录进行比对。比对字段包括：
- 主键
- 关键字段（如姓名、金额、状态）
- 更新时间
如果发现有差异，记录日志，并触发重同步。
代码示例：比对脚本
# 比对脚本：检查 MySQL 和 MongoDB 数据一致性
import pymysql
from pymongo import MongoClient
import random
mysql_conn = pymysql.connect(host='localhost', user='root', password='123456', db='test_db')
mongo_client = MongoClient('mongodb://localhost:27017/')
mongo_db = mongo_client['test_db']
def compare():
cursor = mysql_conn.cursor()
cursor.execute("SELECT id, name, status, update_time FROM users LIMIT 1000")
mysql_records = cursor.fetchall()
# 随机打乱，模拟真实场景
random.shuffle(mysql_records)
errors = []
for row in mysql_records:
user_id = row[0]
name = row[1]
status = row[2]
update_time = row[3]
# 查询 MongoDB
mongo_doc = mongo_db.users.find_one({'_id': user_id})
if not mongo_doc:
errors.append(f"ID {user_id} 在 MongoDB 中不存在")
continue
# 比对关键字段
if mongo_doc.get('name') != name:
errors.append(f"ID {user_id} 名称不一致: MySQL={name}, MongoDB={mongo_doc.get('name')}")
if str(mongo_doc.get('status')) != str(status):
errors.append(f"ID {user_id} 状态不一致: MySQL={status}, MongoDB={mongo_doc.get('status')}")
# 时间比对，允许 1 秒误差
if abs((mongo_doc.get('update_time') - update_time).total_seconds()) > 1:
errors.append(f"ID {user_id} 更新时间误差过大")
if errors:
print(f"发现 {len(errors)} 处不一致:")
for err in errors[:10]: # 只打印前 10 个
print(err)
else:
print("数据一致性检查通过！")
if __name__ == '__main__':
compare()
步骤五：灰度切流
数据比对通过之后，我们可以开始灰度切流。
什么是灰度切流？ 不是把所有流量都切到 MongoDB，而是先切一小部分用户，比如 1%。
如何切？ 可以通过用户 ID 取模，或者通过用户标签。
// 灰度切流逻辑
public User getUserById(Long userId) {
// 灰度判断：user_id 尾数为 0 或 1 的用户，走 MongoDB
boolean graySwitch = (userId % 100) < 1;
if (graySwitch) {
// 从 MongoDB 读
return mongoRepository.findById(userId);
} else {
// 从 MySQL 读
return mysqlRepository.findById(userId);
}
}
切流后，密切监控错误率和性能指标。如果没有问题，逐步扩大灰度比例，直到 100%。
步骤六：停止双写，下线 MySQL 只读
当所有流量都切到 MongoDB 后，我们可以停止双写。但注意，不要立刻删除 MySQL。
建议保留 MySQL 作为冷备份，运行至少一个月。如果 MongoDB 出现问题，可以随时回滚。
第四阶段：常见报错与解决方案
迁移过程中，报错是难免的。我给你列举几个最头疼的问题，以及我们的解决方案。
4.1 报错：The _id field cannot be changed from {old: value} to {new: value}
原因：你在更新文档时，试图修改 _id 字段。
场景：MySQL 的主键是自增 ID，迁移到 MongoDB 后，你可能想把 MySQL 的 ID 作为 MongoDB 的 _id。但在更新时，如果 MySQL 的 ID 变了（比如你误操作），或者你尝试用新的 ID 替换旧的，就会报这个错。
解决方案：
- MongoDB 的
_id是不可变的。如果你需要更改主键，必须先delete，再insert。 - 在迁移脚本中，确保
_id一旦设置，就不再修改。如果 MySQL 有 ID 变更的需求，考虑在应用层处理，而不是改数据库主键。
4.2 报错：Exceeded memory limit for $group
原因：你的聚合查询中，$group 阶段的内存使用超过了 MongoDB 默认限制（100MB）。
场景：你试图在一个大集合上，按某个字段分组并求和，数据量非常大。
解决方案：
- 使用
$facet或分片：如果数据量太大，考虑把数据分片存储，或者使用$facet分阶段处理。 - 增加内存限制：在 MongoDB 配置中，适当调大
maxTimeMS或allowDiskUse。
// 允许使用磁盘，解决内存不足问题
db.collection.aggregate([
{ $group: { _id: "$category", total: { $sum: "$amount" } } },
{ $sort: { total: -1 } }
], { allowDiskUse: true })
4.3 报错：Cannot index secondary key
原因：你尝试在一个嵌套文档的字段上创建索引，但路径不正确。
场景：你的文档结构是：
{
"name": "zhangsan",
"address": {
"city": "Beijing"
}
}
你想在 address.city 上创建索引，但写成了 address.city.index，或者字段名拼写错误。
解决方案：
- 仔细检查字段路径。在 MongoDB 中，嵌套字段的索引路径是用点号连接的，如
address.city。 - 使用
db.collection.getIndexes()查看所有已创建的索引，确认路径是否正确。
4.4 报错：WriteConcernError: E11000 duplicate key error collection
原因：你插入的文档 _id 或唯一索引字段重复了。
场景：双写阶段，MySQL 和 MongoDB 都有数据，你再次插入一条已存在的记录，但 MongoDB 里有唯一索引约束。
解决方案：
- 在插入前，先
findOne检查