这是 MongoDB 分片集群写入路径系列的第 1/6 篇。
本文回答的问题：向分片集合以 ORDERED 方式发送 1,000 条 bulk 写入时，mongos 与 shard 之间会发生多少次往返？为什么只要一个 shard 变慢，整个 bulk 就会变慢？改成 UNORDERED 能得到什么，又要付出什么代价？
基准版本：MongoDB 4.2（源码r4.2.12标签，4.2 官方文档）。4.2 已于 2023-04-30 结束支持（EOL）。4.4 及之后版本的 mongos 行为未经确认（截至 2026-11）。
收到 ORDERED bulk 的 mongos 只把发往同一 shard 的连续区间作为一个批次发送，收到该批次的响应后才生成下一个批次。hashed 分片键会把 bulk 中相邻的文档分散到多个 shard，因此批次几乎每个文档都会被切断。文档 n 条、shard S 个时，期望往返次数为 1 + (n − 1)(1 − 1/S)。把 1,000 条发往 5 个 shard，约为 800 次。不依赖顺序的 bulk 改用 BulkMode.UNORDERED（驱动选项 ordered(false)）后，mongos 会把各 shard 的批次一次性并发发送。
bulk 写入是把多条 insert·update·replace·delete 装进一个请求发送的方式。在 shell 中使用 db.collection.bulkWrite()，在 Spring Data MongoDB 中使用 BulkOperations。客户端发送一次请求，接收一次响应。
bulk 有两种模式。4.2 官方文档 "Ordered vs Unordered Operations" 一节的说明如下。
| 模式 | 执行方式 | 中途出错时 |
|---|---|---|
| ordered（默认值） | "MongoDB executes the operations serially" | 不处理剩余操作，直接返回 |
| unordered | 可以并行执行（不保证） | 继续处理剩余操作 |
在分片集群中，客户端连接到 mongos（路由器）。mongos 根据 bulk 中每条写入的分片键决定它发往哪个 shard（targeting），然后把发往同一 shard 的写入聚在一起，为每个 shard 生成子批次（child batch）并发送。
从客户端看只有一次请求。但在 mongos 与 shard 之间，会产生与子批次数量相同的往返。本文所说的往返，指的就是这种 mongos ↔ shard 往返。
hashed 分片键按字段值的哈希值划分 chunk（按分片键值范围划分的数据单元）。4.2 官方文档 "Hashed Sharding" 这样写道。
Post-hash, documents with "close" shard key values are unlikely to be on the same chunk or shard
例如，设想一个以 _id: { productId, storeId } 作为 hashed 分片键的 store_inventory 集合。这是把一个商品按门店数量展开成多份文档的集合。同一商品的门店 1 文档与门店 2 文档键值相近，但哈希后很可能位于不同的 shard。这是 hashed 分片键为了均匀分布数据而付出的代价。
mongos 在 src/mongo/s/write_ops/batch_write_op.cpp 的 BatchWriteOp::targetBatch 中生成子批次。第 263-292 行的注释这样描述 ordered 规则。
// Targeting of ordered batches is a bit more complex - to respect the ordering of the
// batch, we can only send:
// A) a single targeted batch to one shard endpoint
// B) multiple targeted batches, but only containing targeted writes for a single write op
在一条写入只发往一个 shard 的通常情况下，适用 A。ordered 模式下一次只向一个 shard 发送批次。如果下一条写入发往当前批次之外的 shard，isNewBatchRequiredOrdered（第 96-106 行）就返回 true，mongos 随即切断批次。
执行循环在 batch_write_exec.cpp 中。mongos 在 while (!batchOp.isFinished() && !abortBatch)（第 120 行）内生成并发送批次，然后用 ars.next() 等待响应。第 255 行的注释是 "Block until a response is available."。必须收齐响应才会生成下一个批次。本文把这个循环从发送批次到收齐响应走完一次称为一轮。
所以在 ordered 模式下，发往同一 shard 的一个连续区间就是一次往返。
unordered 模式会对剩余的全部写入做 targeting，每个 shard 打成一个批次。一轮最多并发发送 S 个（S 为 shard 数）子批次。切分批次的条件有三个。
kMaxWriteBatchSize（100,000 条，write_ops.h:62）时切分BSONObjMaxUserSize）时切分isNewBatchRequiredUnordered，第 112-125 行）1,000 条的 bulk 通常不会触发这些条件，一轮即可结束。
假设有 A·B·C 三个 shard，8 条文档按下表分配。分配是示例值。
| 顺序 | 文档 | shard |
|---|---|---|
| 1 | d1 | A |
| 2 | d2 | B |
| 3 | d3 | B |
| 4 | d4 | C |
| 5 | d5 | A |
| 6 | d6 | C |
| 7 | d7 | C |
| 8 | d8 | B |
两种模式生成的批次如下。
| 轮次 | ORDERED | UNORDERED |
|---|---|---|
| 1 | A: [d1] | A: [d1, d5], B: [d2, d3, d8], C: [d4, d6, d7]（并发发送） |
| 2 | B: [d2, d3] | 无 |
| 3 | C: [d4] | 无 |
| 4 | A: [d5] | 无 |
| 5 | C: [d6, d7] | 无 |
| 6 | B: [d8] | 无 |
ORDERED 依次走 6 轮，每一轮都要等待上一轮的响应。UNORDERED 在 1 轮内并发发送 3 个子批次。
ORDERED 的往返次数等于批次被切断的次数加 1。当相邻两条文档的 shard 不同时，批次就会被切断。上例中 7 处相邻对里有 5 处 shard 发生了变化，所以是 1 + 5 = 6 轮。
假设 hashed 分片键让文档均匀且相互独立地分配到 S 个 shard。那么相邻两条文档位于同一 shard 的概率为 1/S，位于不同 shard 的概率为 1 − 1/S。1,000 条文档之间有 999 处边界。shard 为 5 个时，其中 4/5 即约 799 处会切断批次。再加上第一个批次的 1 次，约为 800 次。
通式如下。
기대 왕복 수 = 1 + (n − 1) × (1 − 1/S)
n = 1,000, S = 5 → 1 + 999 × 0.8 = 800.2
按 shard 数计算的值如下（n = 1,000）。
| shard 数 S | ORDERED 期望往返次数 | UNORDERED 轮数（并发请求数） |
|---|---|---|
| 2 | 500.5 | 1 (2) |
| 3 | 667 | 1 (3) |
| 4 | 750.25 | 1 (4) |
| 5 | 800.2 | 1 (5) |
| 6 | 833.5 | 1 (6) |
| 8 | 875.125 | 1 (8) |
shard 越多，ORDERED 往返次数越接近 n。为了提升写入吞吐量而增加 shard，单个 ORDERED bulk 反而会被切得更碎。
write ticket 是限制可同时进入存储引擎的写入数量的入场券。shard 的 mongod 执行写入时以 IX（intent exclusive）模式获取全局锁，此时领取 write ticket。这一行为在 lock_state.cpp 中。获取全局锁的函数 _lockGlobalBegin 会先调用 _acquireTicket。子批次只在该 shard 上执行期间持有 ticket。mongos 收到响应、生成下一个批次的间隙里并不持有。
所以 ORDERED bulk 并不会长时间持有 ticket。但单个 bulk 的生命周期会横跨所有 shard。只要一个 shard 变慢，整个 bulk 就会停滞。5 个 shard 时，一个 bulk 平均经过每个 shard 160 次（800 ÷ 5）。如果其中一个 shard 的响应慢 10 倍，bulk 就要依次等待这个慢响应 160 次。
根据利特尔法则，系统中同时在途的作业数等于每秒到达的作业数乘以单个作业的停留时间。下面的数字全部是示例值。
| 情况 | 单个 bulk 的时间 | 同时在途的 bulk 数 |
|---|---|---|
| ORDERED，正常 | 800 × 1ms + 1,000 × 0.2ms = 1.0 秒 | 20 |
| ORDERED，一个 shard 变慢 | 约 2.8 秒 | 约 56 |
| UNORDERED，正常 | 1ms + 200 × 0.2ms = 41ms | 约 0.8 |
| UNORDERED，一个 shard 变慢 | 10ms + 200 × 2ms = 410ms | 约 8 |
ORDERED·变慢一行的计算如下。发往慢 shard 的 160 轮和 200 条文档耗用 1.6 秒 + 0.4 秒 = 2.0 秒。其余 640 轮和 800 条文档耗用 0.64 秒 + 0.16 秒 = 0.8 秒。
UNORDERED 也要等待最慢的 shard，但这种等待只需付出一次。
同时在途的 bulk 越多，被占住的客户端线程和连接也就越多。如果发送 bulk 的 writer 服务是 Kafka 消费者，那么消费者线程在这段时间内无法处理下一条消息。
data class StoreInventory(
@Id val id: InventoryId, // { productId, storeId }, hashed 샤드키
val stock: Int,
val price: Long,
)
fun saveAll(items: List<StoreInventory>) {
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, StoreInventory::class.java)
items.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
ops.execute()
}
fun saveAll(items: List<StoreInventory>) {
// 같은 _id가 두 번 들어가면 순서가 결과를 바꾼다. 마지막 값만 남긴다.
val deduped = items.associateBy { it.id }.values.toList()
val ops = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, StoreInventory::class.java)
deduped.forEach { item ->
ops.replaceOne(
Query.query(Criteria.where("_id").`is`(item.id)),
item,
FindAndReplaceOptions.options().upsert(),
)
}
try {
ops.execute()
} catch (e: BulkOperationException) {
val failed = e.errors.map { deduped[it.index] }
retryOrDeadLetter(failed)
}
}
如果每个 replaceOne 操作的是不同的文档，执行顺序就不会改变结果。因此两段代码会得到相同的最终状态。顺序依赖产生于同一个 bulk 中放入同一文档两次的情况。UNORDERED 代码在构建 bulk 之前按 _id 只保留最后一个值，从而消除了这种情况。
改用 UNORDERED 的代价是要处理部分失败。ORDERED 在第一个错误处停下，所以处理到哪里由一个位置确定。UNORDERED 出错后仍会继续处理其余写入，因此失败的写入可能散落在 bulk 各处。BulkOperationException.errors 的每一项都有 index，表示该写入在 bulk 中的位置。只挑出失败位置的文档重新处理即可。
直接使用官方 Java 驱动时，改一行选项即可。
collection.bulkWrite(models, BulkWriteOptions().ordered(false))
_id 可能重复，在构建 bulk 之前先合并。index，或发往单独的队列。db.collection.getShardDistribution() 和 sh.status() 查看。要查看某个键发往哪个 shard，就在用该键查询的 find() 后加上 explain()，结果中 winningPlan.shards 列出的 shard 名称就是该键所在的 shard。从 4.0 起还可以用 convertShardKeyToHashed() 查看哈希值。陷阱
- Spring Data MongoDB
bulkOps与 shellbulkWrite()的默认模式是 ordered。如果没有写明模式，就是在以 ORDERED 运行。- UNORDERED 也要等待最慢的 shard。它只是把慢 shard 的影响减少到一次，并不能消除。
- 在 UNORDERED 下，如果一个 bulk 中同一文档出现两次，无法预期哪条写入最后生效。
- 往返发生在 mongos 与 shard 之间，所以只看客户端侧的请求数是看不出来的。
这是未经实际运行验证的步骤。下面的结果是基于源码的预期。
mlaunch init --replicaset --nodes 1 --sharded 5。也可以用 docker compose 搭建。sh.enableSharding("demo")，再执行 sh.shardCollection("demo.store_inventory", { _id: "hashed" })。对空集合做 hashed 分片时，chunk 会预先分到各个 shard。db.serverStatus().metrics.commands.update.total 的值。db.store_inventory.bulkWrite(ops, { ordered: true }) 发送 1,000 条 _id 互不相同的 replace(upsert)。{ ordered: false } 发送相同的 bulk，重复第 3~5 步。预期（基于源码）结果如下。ORDERED 的增量之和落在上面计算表的值附近。UNORDERED 的增量之和等于 shard 数。这是假设一个子批次以一条 update 命令到达 shard 时的预期。
- 实际 chunk 分布：上面的公式假设文档均匀且独立地分配到所有 shard。如果 chunk 集中在某个 shard，同一 shard 连续出现的概率变大，往返次数会减少。分布倾斜到什么程度会减少多少，没有另行计算。
- 4.4 及之后版本的 mongos：依据是 4.2 源码（
r4.2.12）和 4.2 文档。4.4 之后 mongos 的批次规则是否相同，未经确认。- 复现步骤中的计数器：
metrics.commands.update.total是否对每个子批次恰好加 1，profiler 是按子批次还是按单条写入（statement）记录，均未通过实际运行确认。- 代码示例：Kotlin 示例基于 Spring Data MongoDB
BulkOperationsAPI 编写，但没有运行过。