MongoDB 从 4.0 开始支持 replica set 级别的多文档事务，4.2 扩展到分片集群事务。虽然 MongoDB 的事务不如关系型数据库成熟，但在需要强一致性的场景下非常有用。理解其底层隔离机制、驱动层面的使用方式以及生产环境中的最佳实践，对于构建高可靠的数据层至关重要。
1. ACID 保证
| ACID | MongoDB 支持 | 说明 |
|---|---|---|
| Atomicity | ✅ | 事务内所有操作同时成功或同时回滚 |
| Consistency | ✅ | 遵守文档验证规则和索引约束 |
| Isolation | ✅ Snapshot | 快照隔离（Snapshot Isolation） |
| Durability | ✅（可配置） | journal 持久化，可配置 writeConcern |
1.1 Snapshot Isolation 与 Repeatable Read 的区别
传统关系型数据库（如 SQL Server、Oracle）在 Repeatable Read 隔离级别下，事务读取的数据会在事务持续期间被锁定，防止其他事务修改已读数据。然而 Repeatable Read 仍然允许幻读（Phantom Read），因为只锁定已存在的行，不会阻止新行的插入。
MongoDB 采用的是 Snapshot Isolation（快照隔离），这是比 Repeatable Read 更强的隔离级别。快照隔离的核心思想是：每个事务在开始时会获取一个一致性的快照，事务内所有的读取操作都基于这个快照进行，因此天然避免了幻读。由于读取不获取任何锁，读写之间不会相互阻塞。
两者的关键差异在于并发控制策略：Repeatable Read 通过锁机制阻止其他事务修改已读数据，而 Snapshot Isolation 基于多版本并发控制（MVCC），读取操作不会阻塞写入操作，写入操作也不会阻塞读取操作。
1.2 写偏（Write Skew）现象
Snapshot Isolation 虽然避免了幻读和不可重复读，但存在一种称为 写偏（Write Skew） 的异常。写偏发生在两个事务并发读取同一组数据，然后各自修改其中不同的子集，导致业务约束被违反。
典型场景：医院值班系统中，要求每个班次至少有一名医生值班。两个医生 A 和 B 同时查看班次表，发现都有人在值班，于是各自申请休假。由于每个人修改的是不同文档，快照隔离下两个事务都能成功提交，导致该班次无人值班。
1.3 MongoDB 冲突解决策略
MongoDB 在事务提交时执行冲突检测。如果在事务持有快照期间，另一个事务修改了相同的文档并先提交，当前事务提交时会检测到写冲突，抛出 WriteConflict 异常并自动中止。
// Mongo Shell: 模拟写冲突检测
const session1 = db.getMongo().startSession();
const session2 = db.getMongo().startSession();
session1.startTransaction({readConcern: {level: "snapshot"}});
session2.startTransaction({readConcern: {level: "snapshot"}});
// 两个事务同时读取同一文档
const doc1 = session1.getDatabase("shop").inventory.findOne({sku: "SKU001"});
const doc2 = session2.getDatabase("shop").inventory.findOne({sku: "SKU001"});
// session1 先提交修改
session1.getDatabase("shop").inventory.updateOne(
{sku: "SKU001"}, {$inc: {qty: -1}}, {session: session1}
);
session1.commitTransaction();
// session2 尝试提交时会因写冲突失败
try {
session2.getDatabase("shop").inventory.updateOne(
{sku: "SKU001"}, {$inc: {qty: -1}}, {session: session2}
);
session2.commitTransaction(); // 抛出 WriteConflict
} catch (e) {
print("WriteConflict detected: " + e.message);
session2.abortTransaction();
} finally {
session1.endSession();
session2.endSession();
}
解决写偏的推荐方案是在相关文档上使用显式的文档级锁或引入版本号字段，确保并发修改时能够检测冲突。
2. 事务基础用法
2.1 Mongo Shell
const session = db.getMongo().startSession();
session.startTransaction({
readConcern: { level: "snapshot" },
writeConcern: { w: "majority" }
});
try {
const orders = session.getDatabase("shop").orders;
const inventory = session.getDatabase("shop").inventory;
// 扣减库存
inventory.updateOne(
{ sku: "SKU001" },
{ $inc: { qty: -2 } },
{ session }
);
// 创建订单
orders.insertOne({
userId: "u123",
items: [{ sku: "SKU001", qty: 2 }],
status: "paid",
createdAt: new Date()
}, { session });
session.commitTransaction();
} catch (error) {
session.abortTransaction();
throw error;
} finally {
session.endSession();
}
2.2 驱动程序（Java）
MongoClient client = MongoClients.create("mongodb://localhost:27017");
ClientSession session = client.startSession();
TransactionOptions txnOptions = TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY)
.readPreference(ReadPreference.primary())
.build();
TransactionBody<String> txnBody = () -> {
MongoDatabase db = client.getDatabase("shop");
db.getCollection("inventory").updateOne(
session, Filters.eq("sku", "SKU001"),
Updates.inc("qty", -2)
);
db.getCollection("orders").insertOne(session,
new Document("userId", "u123")
.append("status", "paid")
.append("createdAt", new Date())
);
return "Order created successfully";
};
try {
String result = session.withTransaction(txnBody, txnOptions);
System.out.println(result);
} catch (RuntimeException e) {
// 事务自动回滚
throw e;
} finally {
session.close();
}
2.3 Node.js 驱动
const { MongoClient } = require('mongodb');
async function runTransaction() {
const client = new MongoClient('mongodb://localhost:27017');
await client.connect();
const session = client.startSession();
const txnOptions = {
readConcern: { level: 'snapshot' },
writeConcern: { w: 'majority' },
readPreference: 'primary'
};
try {
await session.withTransaction(async () => {
const shop = client.db('shop');
await shop.collection('inventory').updateOne(
{ sku: 'SKU001' },
{ $inc: { qty: -2 } },
{ session }
);
await shop.collection('orders').insertOne({
userId: 'u123',
items: [{ sku: 'SKU001', qty: 2 }],
status: 'paid',
createdAt: new Date()
}, { session });
}, txnOptions);
console.log('Transaction committed');
} catch (error) {
console.error('Transaction aborted:', error);
throw error;
} finally {
await session.endSession();
await client.close();
}
}
runTransaction();
2.4 Python（PyMongo）
from pymongo import MongoClient
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern
client = MongoClient("mongodb://localhost:27017")
with client.start_session() as session:
with session.start_transaction(
read_concern=ReadConcern('snapshot'),
write_concern=WriteConcern('majority')
):
shop = client['shop']
shop.inventory.update_one(
{'sku': 'SKU001'},
{'$inc': {'qty': -2}},
session=session
)
shop.orders.insert_one({
'userId': 'u123',
'items': [{'sku': 'SKU001', 'qty': 2}],
'status': 'paid',
'createdAt': datetime.utcnow()
}, session=session)
session.commit_transaction()
print("Transaction committed")
2.5 Spring Data MongoDB
@Service
public class OrderService {
@Autowired
private MongoTemplate mongoTemplate;
@Transactional
public void createOrder(String userId, String sku, int qty) {
// 扣减库存
mongoTemplate.updateFirst(
Query.query(Criteria.where("sku").is(sku)),
new Update().inc("qty", -qty),
Inventory.class
);
// 创建订单
Order order = new Order();
order.setUserId(userId);
order.setItems(List.of(new OrderItem(sku, qty)));
order.setStatus("paid");
order.setCreatedAt(new Date());
mongoTemplate.save(order);
}
}
Spring Data MongoDB 的 @Transactional 注解会自动管理会话生命周期。默认情况下，事务使用 MongoTransactionManager 进行协调，异常会自动触发回滚。
2.6 事务嵌套与 Savepoint 概念
MongoDB 不支持事务嵌套。在一个已开启事务的会话中再次调用 startTransaction() 会抛出异常。如果业务逻辑需要分阶段提交，应用层应采用补偿机制或 Saga 模式来模拟嵌套语义。
// Java: 检测嵌套事务并抛出异常
try {
session.startTransaction();
// 以下调用会抛出 IllegalStateException
session.startTransaction(); // MongoDB 不支持嵌套事务
} catch (IllegalStateException e) {
System.err.println("Nested transactions not supported: " + e.getMessage());
}
虽然 MongoDB 没有原生的 Savepoint，但在 WiredTiger 存储引擎层面，可以通过维护多个会话快照间接实现阶段性回退。在应用层，建议使用预写入（write-ahead log）或事件溯源（Event Sourcing）来记录操作历史，以便手动执行补偿回滚。
// Node.js: 应用层模拟 Savepoint 的补偿逻辑
async function executeWithSavepoint(session, operations) {
const compensations = [];
for (const op of operations) {
try {
const undo = await op.execute(session);
compensations.push(undo);
} catch (error) {
// 反向执行补偿操作
for (const compensate of compensations.reverse()) {
await compensate(session);
}
throw error;
}
}
}
3. 事务限制
| 限制 | 说明 |
|---|---|
| 事务超时 | 默认 60 秒（maxTransactionLockRequestTimeoutMillis），可调整为 10 秒至 10 分钟 |
| 文档修改 | 事务内无法创建/删除集合、索引，无法执行 createCollection、createIndex、drop 等命令 |
| 操作数量 | 无硬性限制，但受 oplog 大小限制（约 16MB BSON 文档限制） |
| 分片事务 | 涉及多个分片时性能下降明显，4.2 起支持 |
| 会话 | 每个事务必须绑定一个会话，会话有 30 分钟默认空闲超时 |
3.1 事务中无法执行的完整操作列表
事务中无法执行的操作包括但不限于：
- 集合与索引管理：
createCollection、createIndexes、drop、dropIndexes、collMod、compact - 用户与角色管理：
createUser、dropUser、createRole、dropRole - 分片操作：
shardCollection、moveChunk、splitChunk、balancerStart、balancerStop - 某些查询命令：
count（推荐使用countDocuments或聚合管道）、distinct（部分驱动支持有限）、geoSearch - 诊断命令：
profile、setParam $out和$merge阶段的聚合管道（这些操作会写入新集合）
// Mongo Shell: 事务中尝试创建集合将失败
const session = db.getMongo().startSession();
session.startTransaction();
try {
session.getDatabase("shop").createCollection("temp_orders"); // 抛出错误
session.commitTransaction();
} catch (e) {
print("Error: " + e.message); // Cannot run createCollection in a multi-document transaction
session.abortTransaction();
} finally {
session.endSession();
}
3.2 事务大小限制
事务中的批量操作累积的 oplog 条目不得超过 16MB。这是由于 MongoDB 的 BSON 文档大小限制决定的。如果事务涉及大量文档更新或插入，oplog 条目可能超出此限制，导致事务失败。
# Python: 大批量事务可能触发 oplog 大小限制
with client.start_session() as session:
with session.start_transaction():
# 以下操作更新了数万个文档，可能超过 16MB oplog 限制
result = client['shop']['inventory'].update_many(
{'status': 'pending'},
{'$set': {'status': 'processed'}},
session=session
)
print(f"Matched: {result.matched_count}")
缓解措施：将大批量操作拆分为多个独立事务，每批控制在数千条文档以内。
3.3 锁超时与会话过期
事务中的写操作会持有文档级锁和集合级意向锁。如果锁请求超时（默认 5 毫秒），事务会被中止。
// Mongo Shell: 查看锁超时配置
db.adminCommand({getParameter: 1, maxTransactionLockRequestTimeoutMillis: 1});
// 默认值为 5（毫秒），可调整至合理范围以避免频繁超时
会话有 30 分钟 的默认空闲超时（localLogicalSessionTimeoutMinutes）。如果事务持续时间过长或会话长时间不活动，服务器会自动终止会话并回滚未提交的事务。
// Java: 捕获会话过期异常
try {
session.withTransaction(txnBody, txnOptions);
} catch (MongoTimeoutException e) {
System.err.println("Session expired, transaction rolled back: " + e.getMessage());
}
4. 性能影响
单文档写入: 50,000+ ops/s
多文档事务: 5,000-10,000 ops/s （事务开销约 5-10x）
事务延迟来源:
- 获取快照（Snapshot）
- 两阶段提交协调（分片事务）
- oplog 同步
4.1 WiredTiger 快照管理
MongoDB 使用 WiredTiger 存储引擎提供的 MVCC 机制实现快照隔离。每个事务开始时，WiredTiger 记录当前的 snapshot_timestamp，事务内所有读取都基于此时间戳的可见性规则。
WiredTiger 维护一个全局事务表，记录所有活跃事务的 ID。旧版本数据不会被立即清理，而是保留到没有更早的快照需要读取它为止。这意味着长时间运行的事务会导致缓存中存在大量旧版本数据，增加内存压力。
// Mongo Shell: 查看当前活跃事务和快照信息
// WiredTiger 内部通过 WT_SESSION 管理事务状态
// 可通过 serverStatus 观察缓存压力
db.serverStatus().wiredTiger.cache;
4.2 事务对 oplog 的影响
多文档事务中的所有操作会被打包写入一条 oplog 条目（applyOps 命令），这保证了副本集节点能够原子性重放事务。打包写入虽然保证了一致性，但也带来了额外开销：
- oplog 大小膨胀：事务内所有操作被序列化为单个大文档
- 副本集同步延迟：备节点需要完整接收并解析整个事务后才能应用
- 触发 oplog 中的
commitTransaction条目：包含最终提交时间戳
// Mongo Shell: 查看 oplog 中的事务条目
use local;
db.oplog.rs.find({"op": "c", "o.applyOps": {$exists: true}}).sort({ts: -1}).limit(5);
高并发事务场景下，oplog 的写入带宽可能成为瓶颈，建议监控 repl.buffer.count 和 repl.buffer.maxSizeBytes。
4.3 内存压力与事务超时关系
WiredTiger 缓存默认占用系统物理内存的 50%（不超过 256GB）。长时间运行的事务会阻止 WiredTiger 清理旧版本数据，导致缓存命中率下降，进而触发频繁的磁盘读取。
当缓存压力达到临界点时，系统会主动中止部分长时间运行的事务以释放资源。因此，事务超时不仅是配置参数的问题，还与系统整体的内存和缓存健康度密切相关。
# mongod.conf: 调整 WiredTiger 缓存和事务超时
storage:
wiredTiger:
engineConfig:
cacheSizeGB: 4
setParameter:
maxTransactionLockRequestTimeoutMillis: 10
最佳实践：能用单文档原子操作（update + $inc 等）就不用事务；控制事务持续时间低于 1 秒；限制事务内操作数量在合理范围。
5. 分布式事务：两阶段提交在分片集群中的实现
MongoDB 4.2 引入了对分片集群的分布式事务支持。在分片环境中，事务可能涉及多个分片上的文档，此时 MongoDB 使用 两阶段提交（Two-Phase Commit, 2PC） 协议来保证原子性。
5.1 两阶段提交流程
第一阶段（Prepare Phase）：协调器（mongos 或 primary 分片）向所有参与分片发送 Prepare 请求，各分片在本地执行操作并将结果记录在本地 oplog，但不最终提交。
第二阶段（Commit Phase）：如果所有分片都返回成功，协调器发送 Commit 请求，各分片将本地事务标记为已提交；如果有任何一个分片失败，协调器发送 Abort 请求回滚所有分片的本地事务。
MongoDB 在分片集群中的两阶段提交由 Transaction Coordinator 组件自动管理。协调器通常在首个参与分片的主节点上运行，使用 config 数据库中的 transaction_coordinators 集合持久化协调状态，避免因协调器崩溃导致事务悬挂。
5.2 分片事务代码示例
// Node.js: 跨分片事务（基于分片键路由到不同分片）
const { MongoClient } = require('mongodb');
async function crossShardTransaction() {
const client = new MongoClient('mongodb://mongos1:27017,mongos2:27017');
await client.connect();
const session = client.startSession();
try {
await session.withTransaction(async () => {
const shop = client.db('shop');
// 假设 orders 按 region 分片，以下可能涉及不同分片
await shop.collection('orders').insertOne({
orderId: 'O1001',
region: 'apac',
amount: 299,
status: 'paid'
}, { session });
await shop.collection('orders').insertOne({
orderId: 'O1002',
region: 'emea',
amount: 150,
status: 'paid'
}, { session });
// 更新汇总表（可能位于独立分片）
await shop.collection('daily_summary').updateOne(
{ date: new Date().toISOString().split('T')[0] },
{ $inc: { totalOrders: 2, totalAmount: 449 } },
{ upsert: true, session }
);
});
console.log('Cross-shard transaction committed');
} catch (error) {
console.error('Cross-shard transaction failed:', error);
throw error;
} finally {
await session.endSession();
await client.close();
}
}
crossShardTransaction();
// Java: 分片集群事务配置
MongoClient client = MongoClients.create(
MongoClientSettings.builder()
.applyToClusterSettings(builder ->
builder.hosts(Arrays.asList(
new ServerAddress("mongos1", 27017),
new ServerAddress("mongos2", 27017)
))
)
.build()
);
TransactionOptions txnOptions = TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY)
.build();
// 分片事务调用方式与副本集事务完全一致
// 底层会自动触发 2PC 协调
5.3 分片事务的性能考量
分片事务相比副本集事务有显著的性能开销：
- 网络往返增加：每个分片至少需 2 轮 Prepare + Commit 通信
- 协调器持久化：协调状态写入 config 服务器增加了额外延迟
- 锁粒度扩大：涉及的分片越多，锁冲突概率越高
建议：将相关数据通过分片策略（zone sharding）尽量集中在同一分片上，减少跨分片事务的发生频率。
6. 事务与 Change Streams、触发器的设计冲突与解决
6.1 Change Streams 的事务可见性
MongoDB Change Streams 默认订阅的是 已提交事务 的变更。当事务提交后，Change Streams 会将事务内的所有操作以原子性批次的形式推送，确保消费者看到的是一致的最终状态。
// Node.js: 监听包含事务的 Change Stream
const changeStream = client.db('shop').watch([
{ $match: { 'fullDocument.status': 'paid' } }
], { fullDocument: 'updateLookup' });
changeStream.on('change', (change) => {
// 如果变更来自事务，change 对象包含 txnNumber 和 lsid
if (change.txnNumber !== undefined) {
console.log('Change from transaction:', change.txnNumber);
}
console.log('Operation type:', change.operationType);
console.log('Full document:', change.fullDocument);
});
6.2 触发器设计冲突
在使用触发器（如 MongoDB Atlas Triggers 或自研触发器服务）时，可能存在以下冲突：
- 重复处理：事务提交后 Change Stream 推送原子批次，触发器可能将同一事务内的多个操作拆分为独立事件处理
- 顺序依赖：触发器处理的顺序可能与事务内操作顺序不一致，导致下游状态错误
- 补偿困难：触发器触发的副作用（如发送邮件、调用 API）难以在事务回滚时撤销
6.3 解决方案
方案一：幂等设计。确保下游处理器对同一事件的多次处理不会产生副作用。引入唯一事件 ID，在下游存储处理记录。
# Python: Change Stream 消费者的幂等处理
with client.watch([{'$match': {'operationType': 'insert'}}]) as stream:
for change in stream:
event_id = change['_id']['_data']
if processed_events_collection.find_one({'_id': event_id}):
continue # 已处理，跳过
process_change(change)
processed_events_collection.insert_one({
'_id': event_id,
'processedAt': datetime.utcnow()
})
方案二：外部 Outbox 模式。不使用 Change Streams 直接触发副作用，而是将需要触发的事件写入一个 outbox 集合，由独立的轮询器读取并执行。这样可以保证事件仅在事务成功提交后才对外可见。
// Java: Outbox 模式的事务实现
@Transactional
public void placeOrder(Order order) {
// 1. 保存订单
mongoTemplate.save(order);
// 2. 写入 outbox 事件（与订单在同一事务中）
OutboxEvent event = new OutboxEvent();
event.setAggregateType("Order");
event.setAggregateId(order.getId());
event.setType("OrderPlaced");
event.setPayload(toJson(order));
event.setCreatedAt(new Date());
mongoTemplate.save(event);
}
方案三：延迟确认。触发器在处理完事件后，延迟一小段时间再确认，利用这段时间窗口检测事务是否提交持久化。
7. 生产最佳实践：事务超时调优、重试策略、监控指标
7.1 事务超时调优
生产环境中需要根据业务负载调整事务超时参数：
# mongod.conf
setParameter:
maxTransactionLockRequestTimeoutMillis: 10 # 锁等待超时（毫秒）
transactionLifetimeLimitSeconds: 60 # 事务总生命周期（秒）
对于 OLTP 型负载，建议将 transactionLifetimeLimitSeconds 设置为 10-30 秒，避免长时间持有快照和锁。对于批处理型事务，可适当放宽至 120 秒。
// Mongo Shell: 动态调整事务超时
db.adminCommand({
setParameter: 1,
transactionLifetimeLimitSeconds: 30
});
7.2 重试策略
驱动级别的重试逻辑应区分可重试错误和不可重试错误：
| 错误类型 | 是否可重试 | 处理策略 |
|---|---|---|
| TransientTransactionError | ✅ | 重新开启新会话和事务重试 |
| UnknownTransactionCommitResult | ✅ | 在 Commit 阶段重试提交操作 |
| WriteConflict | ❌ | 业务层决定重试或放弃 |
| LockTimeout | ✅ | 短暂延迟后重试 |
// Java: 完整的可重试事务模式
public <T> T executeWithRetry(TransactionBody<T> txnBody) {
int maxRetries = 3;
for (int i = 0; i < maxRetries; i++) {
try (ClientSession session = client.startSession()) {
return session.withTransaction(txnBody);
} catch (MongoException e) {
if (!isRetryable(e) || i == maxRetries - 1) {
throw e;
}
// 指数退避
try {
Thread.sleep((long) (Math.pow(2, i) * 100));
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException(ie);
}
}
}
throw new IllegalStateException("Should not reach here");
}
private boolean isRetryable(MongoException e) {
return e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL)
|| e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL);
}
# Python: 事务重试包装器
import time
from pymongo.errors import PyMongoError
def with_retry(max_retries=3):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except PyMongoError as e:
if attempt == max_retries - 1:
raise
if 'TransientTransactionError' in str(e) or \
'UnknownTransactionCommitResult' in str(e):
time.sleep(2 ** attempt * 0.1)
else:
raise
return wrapper
return decorator
@with_retry(max_retries=3)
def process_order():
with client.start_session() as session:
with session.start_transaction():
# 事务操作
pass
7.3 监控指标
MongoDB 提供了丰富的事务相关监控指标，通过 db.serverStatus().transactions 可以获取关键数据：
// Mongo Shell: 查看事务统计
db.serverStatus().transactions
关键监控指标：
transactions.currentActive：当前活跃事务数。持续高值可能表明事务运行时间过长transactions.currentInactive：已开启但未执行操作的事务数transactions.currentOpen：当前打开的事务总数（active + inactive）transactions.totalStarted：自启动以来开启的事务总数transactions.totalCommitted：成功提交的事务数transactions.totalAborted：中止的事务数transactions.totalPrepared：参与两阶段提交的 prepared 事务数
# 通过 MongoDB Exporter 导出到 Prometheus
# 关键告警规则示例
# 1. 事务中止率 > 5%
# 2. 活跃事务数 > 100
# 3. 平均事务持续时间 > 5s
# Python: 程序化获取事务指标
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017")
status = client.admin.command('serverStatus')
txns = status.get('transactions', {})
print(f"活跃事务: {txns.get('currentActive', 0)}")
print(f"总提交: {txns.get('totalCommitted', 0)}")
print(f"总中止: {txns.get('totalAborted', 0)}")
abort_rate = txns.get('totalAborted', 0) / max(txns.get('totalStarted', 1), 1)
if abort_rate > 0.05:
print(f"告警: 事务中止率 {abort_rate:.2%} 过高")
8. 替代方案设计：单文档原子操作、乐观锁、补偿事务（Saga 模式）
MongoDB 事务虽然功能完善，但并非所有场景都需要多文档事务。以下替代方案在性能和复杂度之间提供了不同的权衡。
8.1 单文档原子操作
MongoDB 的单文档操作天然具有原子性。通过良好的文档建模，可以将需要同时更新的数据嵌入到同一文档中，完全避免事务开销。
// Mongo Shell: 嵌套文档替代多文档事务
// 将订单项嵌入订单文档，实现原子更新
db.orders.updateOne(
{ _id: ObjectId("..."), "items.sku": "SKU001" },
{
$set: { status: "shipped", updatedAt: new Date() },
$inc: { "items.$.shippedQty": 2 }
}
);
// Java: 利用嵌入式子文档避免事务
mongoTemplate.updateFirst(
Query.query(Criteria.where("_id").is(orderId)),
new Update()
.set("status", "shipped")
.set("updatedAt", new Date())
.inc("items.$.shippedQty", qty),
Order.class
);
适用场景：1:1 或 1:N 关系且子文档数量可控；需要同时更新的字段可以聚合到同一文档。
8.2 乐观锁（_version_）
乐观锁通过版本号机制检测并发修改冲突，避免了事务的锁定开销，适用于读多写少的场景。
// Mongo Shell: 乐观锁实现
function updateWithOptimisticLock(collection, filter, updateOps) {
while (true) {
const doc = collection.findOne(filter);
if (!doc) throw new Error("Document not found");
const result = collection.updateOne(
{ _id: doc._id, _version_: doc._version_ },
{
...updateOps,
$inc: { _version_: 1 }
}
);
if (result.modifiedCount === 1) {
return true; // 更新成功
}
// 版本冲突，重试
}
}
// 使用
updateWithOptimisticLock(
db.inventory,
{ sku: "SKU001" },
{ $inc: { qty: -2 } }
);
# Python: 乐观锁封装
from pymongo import ReturnDocument
def update_with_version(collection, filter, update):
max_retries = 5
for _ in range(max_retries):
doc = collection.find_one(filter)
if not doc:
raise ValueError("Document not found")
version = doc.get('_version_', 0)
result = collection.update_one(
{**filter, '_version_': version},
{**update, '$inc': {'_version_': 1}}
)
if result.modified_count == 1:
return True
raise RuntimeError("Max retries exceeded for optimistic lock")
8.3 补偿事务（Saga 模式）
在微服务架构中，跨服务的分布式事务通常不适合使用数据库级事务。Saga 模式通过一系列本地事务和补偿操作来实现最终一致性。
// Java: Saga 模式伪代码实现
@Component
public class OrderSaga {
@Autowired private InventoryService inventoryService;
@Autowired private PaymentService paymentService;
@Autowired private ShippingService shippingService;
public void executeOrderSaga(OrderRequest request) {
List<Runnable> compensations = new ArrayList<>();
try {
// Step 1: 扣减库存
inventoryService.decrease(request.getSku(), request.getQty());
compensations.add(() -> inventoryService.increase(request.getSku(), request.getQty()));
// Step 2: 扣款
paymentService.charge(request.getUserId(), request.getAmount());
compensations.add(() -> paymentService.refund(request.getUserId(), request.getAmount()));
// Step 3: 创建配送单
shippingService.createShipment(request);
// 最后一步无需补偿，或加入退款补偿
} catch (Exception e) {
// 反向执行补偿
Collections.reverse(compensations);
for (Runnable compensation : compensations) {
try {
compensation.run();
} catch (Exception compError) {
// 记录补偿失败，需人工介入
alertAdmin("Compensation failed", compError);
}
}
throw new SagaFailedException("Order saga failed, compensations executed", e);
}
}
}
// Node.js: Saga 协调器
class Saga {
constructor() {
this.steps = [];
this.compensations = [];
}
addStep(action, compensate) {
this.steps.push(action);
this.compensations.unshift(compensate);
return this;
}
async execute() {
const completed = [];
try {
for (const step of this.steps) {
await step();
completed.push(step);
}
} catch (error) {
for (const compensate of this.compensations) {
try {
await compensate();
} catch (compError) {
console.error('Compensation failed:', compError);
}
}
throw error;
}
}
}
// 使用示例
const orderSaga = new Saga()
.addStep(
() => inventoryService.decrease('SKU001', 2),
() => inventoryService.increase('SKU001', 2)
)
.addStep(
() => paymentService.charge('u123', 299),
() => paymentService.refund('u123', 299)
);
orderSaga.execute()
.then(() => console.log('Saga completed'))
.catch(err => console.error('Saga failed:', err));
总结：单文档原子操作是最优选择；乐观锁适合中等并发场景；Saga 模式适合跨服务的长事务。数据库级多文档事务应作为最后的选择，仅在数据模型无法避免且一致性要求绝对严格时使用。
延伸阅读
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。