企业官网建设流程全解析

7.3 批量操作优化：BulkWrite

BulkWrite（批量写入）是MongoDB提供的核心批量操作接口，能够将插入、更新、删除等多个操作打包成一批发送到服务端，大幅减少客户端与服务端的网络往返次数（Round-Trip），相比循环单条操作可提升数倍甚至数十倍的性能。

• 本节从性能对比、核心实战、百万级数据导入优化三个维度，全面讲解BulkWrite的使用与优化思路。

前置准备

1. 环境要求

• MongoDB版本：3.2+（BulkWrite基础支持），推荐4.4+（性能更优）
• 驱动：以Pythonpymongo（3.12+）为例（Node.js/Java驱动语法逻辑一致）
• 安装依赖：pip install pymongo python-dotenv

2. 基础连接代码

frompymongoimportMongoClient,InsertOne,UpdateOne,DeleteOnefrompymongo.errorsimportBulkWriteErrorimporttimeimportrandomimportstring# 1. 连接MongoDB（本地/远程）client=MongoClient("mongodb://localhost:27017/")# 2. 选择数据库和集合db=client["test_db"]collection=db["bulk_demo"]# 清空测试集合（仅测试用）collection.delete_many({})

7.3.1 循环单条操作vs批量操作：性能差异对比

核心差异

实战性能对比（测试10000条插入）

frompymongoimportMongoClient,InsertOne,UpdateOne,DeleteOnefrompymongo.errorsimportBulkWriteErrorimporttimeimportrandomimportstring# 1. 连接MongoDB（本地/远程）client=MongoClient("mongodb://localhost:27017/")# 2. 选择数据库和集合db=client["test_db"]collection=db["bulk_demo"]# 清空测试集合（仅测试用）collection.delete_many({})# 生成测试数据（随机字符串）defgenerate_random_str(length=10):return''.join(random.choice(string.ascii_letters)for_inrange(length))# 测试数据量TEST_COUNT=10000test_data=[{"name":generate_random_str(),"age":random.randint(18,60)}for_inrange(TEST_COUNT)]# ========== 测试1：循环单条插入 ==========start_time=time.time()fordocintest_data:collection.insert_one(doc)single_cost=time.time()-start_timeprint(f"循环单条插入{TEST_COUNT}条耗时：{single_cost:.2f}秒")# 清空集合，准备批量测试collection.delete_many({})# ========== 测试2：BulkWrite批量插入 ==========start_time=time.time()# 构造批量插入操作列表bulk_operations=[InsertOne(doc)fordocintest_data]# 执行批量操作result=collection.bulk_write(bulk_operations)b