文章目录
- 7.3 批量操作优化：BulkWrite
- 前置准备
- 1. 环境要求
- 2. 基础连接代码
- 7.3.1 循环单条操作vs批量操作：性能差异对比
- 核心差异
- 实战性能对比（测试10000条插入）
- 典型输出结果（参考）
- 差异原因分析
- 7.3.2 BulkWrite实战：批量插入、更新、删除组合操作
- 核心语法
- 实战：组合操作完整示例
- 输出结果（参考）
- 关键注意事项
- 7.3.3 场景实战：百万级日志数据批量导入优化
- 业务场景
- 优化思路
- 完整优化代码
- 典型输出结果（参考）
- 进一步优化建议
- 7.3.4 核心总结
- 扩展：不同驱动的BulkWrite语法（参考）
7.3 批量操作优化：BulkWrite
BulkWrite（批量写入）是MongoDB提供的核心批量操作接口，能够将插入、更新、删除等多个操作打包成一批发送到服务端，大幅减少客户端与服务端的网络往返次数（Round-Trip），相比循环单条操作可提升数倍甚至数十倍的性能。
- 本节从性能对比、核心实战、百万级数据导入优化三个维度，全面讲解
BulkWrite的使用与优化思路。
前置准备
1. 环境要求
- MongoDB版本：3.2+（BulkWrite基础支持），推荐4.4+（性能更优）
- 驱动：以Python
pymongo（3.12+）为例（Node.js/Java驱动语法逻辑一致） - 安装依赖：
pip install pymongo python-dotenv
2. 基础连接代码
from pymongo import MongoClient, InsertOne, UpdateOne, DeleteOne
from pymongo.errors import BulkWriteError
import time
import random
import string
# 1. 连接MongoDB（本地/远程）
client = MongoClient("mongodb://localhost:27017/")
# 2. 选择数据库和集合
db = client["test_db"]
collection = db["bulk_demo"]
# 清空测试集合（仅测试用）
collection.delete_many({
})
7.3.1 循环单条操作vs批量操作：性能差异对比
核心差异
| 操作类型 | 网络往返次数 | 性能特点 | 适用场景 |
|---|---|---|---|
| 循环单条操作 | N次（N=操作数） | 网络开销大、吞吐量低 | 少量操作（<100） |
| BulkWrite批量 | 1次（分块时少量） | 网络开销极小、吞吐量高 | 大量操作（≥100） |
实战性能对比（测试10000条插入）
from pymongo import MongoClient, InsertOne, UpdateOne, DeleteOne
from pymongo.errors import BulkWriteError
import time
import random
import string
# 1. 连接MongoDB（本地/远程）
client = MongoClient("mongodb://localhost:27017/")
# 2. 选择数据库和集合
db = client["test_db"]
collection = db["bulk_demo"]
# 清空测试集合（仅测试用）
collection.delete_many({
})
# 生成测试数据（随机字符串）
def generate_random_str(length=10):
return ''.join(random.choice(string.ascii_letters) for _ in range(length))
# 测试数据量
TEST_COUNT = 10000
test_data = [{
"name": generate_random_str(), "age": random.randint(18, 60)} for _ in range(TEST_COUNT)]
# ========== 测试1：循环单条插入 ==========
start_time = time.time()
for doc in test_data:
collection.insert_one(doc)
single_cost = time.time() - start_time
print(f"循环单条插入{
TEST_COUNT}条耗时：{
single_cost:.2f}秒")
# 清空集合，准备批量测试
collection.delete_many({
})
# ========== 测试2：BulkWrite批量插入 ==========
start_time = time.time()
# 构造批量插入操作列表
bulk_operations = [InsertOne(doc) for doc in test_data]
# 执行批量操作
result = collection.bulk_write(bulk_operations)
b