如何使用 MongoDB 实现 SELECT ... FOR UPDATE 和 Transactions 水平扩展服务？

Question

我正在尝试使用 MongoDB 以最佳方式实现悲观锁。

目标

我想 H 扩展一项服务：

1. 从 MongoDB 集合中获取基于限制参数和查询过滤器的数据列表
2. 处理每个数据，对每个获取的文档和其他一些内容进行一些更改
3. 更新数据库中的每个文档

所以理想情况下，只要我能够避免在第 1 步处理相同的文档，我就可以运行 N 个处理 1、2、3 的实例。

测试

我查看了文档并使用了一些基本的代码 sn-ps 并最终得出结论：

• 即使使用事务，单独查找和更新，也不会 防止任何其他事务处理相同的数据。这将是 仅当更新方法已被调用时才阻止，即使 交易尚未提交
• 如果两个事务对同一文档执行更新，并且 第二个事务在第一个事务之前提交，第一个事务的更改 将成为丢弃者。所以在同一个文档上开始的最后一个事务获胜

结论

MongoDB 事务在我的用例中没有用。

我的查找结果不是使用 LIMIT，而是 LIMIT 将是我将执行 findAndUpdate 的次数，以避免从扩展的服务中查找和处理相同的数据

const promises = []

for (let i = 0; i <= LIMIT - 1; i += 1) {
  promises.push(
    db.collection(COLLECTION)
      .findOneAndUpdate(
        { status: 'active', lock: 'NONE' },
        { $set: { lock: 'ACTIVE' } },
        { returnOriginal: false }))
}

const itemsToBeProcessed = await Promise.all(promises)

... stuff like updating each object

... update each document with proper object (setting lock to NONE again)

我做得对吗？有什么建议或资源吗？

Answer 1

与无锁解决方案相比，锁通常会降低可伸缩性，因为在持有锁时，其他操作会被阻塞。

如果您打算使用锁定来保留工作项，则在实际处理项目时（在将其分配给工作人员加工）。您可以使用事务或可能更简单的解决方案（如条件更新）来实现预订本身。另一方面，您通常需要某种机制来检测死亡工人并释放因故障而停止工作的保留。

如果您在整个数据集上使用事务（而不仅仅是预订系统），这将无法扩展，您可能需要重新构建处理代码。