MongoDB4.0下数组元素Int64字段增量超过Long.MAX_VALUE的实现方案

可行实现方案 #

方案前提 #

MongoDB 4.0 版本仅副本集部署支持多文档事务，若你使用单节点部署，可临时切换为单节点副本集开启事务能力，或使用无事务的逐文档原子更新方案。

方案1：副本集事务方案（推荐，满足全局事务性要求） #

该方案在事务内完成全量数据更新，要么全部生效要么全部回滚，事务修改文档时会持有文档级排他锁，提交前其他读写操作都会阻塞，不会出现中间状态被读取或者其他写入干扰的问题，完全匹配需求。

执行脚本

// 替换为实际的数据库名、集合名
const DB_NAME = "your_database";
const COLL_NAME = "your_collection";
// 增量值为 Long.MAX_VALUE + 1 = 2^63
const DELTA = NumberLong("9223372036854775808");

const session = db.getMongo().startSession({readPreference: "primary"});
const coll = session.getDatabase(DB_NAME).getCollection(COLL_NAME);

try {
    // 开启事务，使用快照读保证读取一致性，多数派写保证数据持久化
    session.startTransaction({
        readConcern: {level: "snapshot"},
        writeConcern: {w: "majority"}
    });

    const cursor = coll.find({"versions.revision": {$lt: 0}}).noCursorTimeout();
    while (cursor.hasNext()) {
        const doc = cursor.next();
        // 应用层计算新的versions数组，避免数据库端运算的溢出、版本兼容问题
        const newVersions = doc.versions.map(elem => {
            if (elem.revision.lt(0)) {
                return Object.assign({}, elem, {revision: elem.revision.add(DELTA)});
            }
            return elem;
        });
        coll.updateOne({_id: doc._id}, {$set: {versions: newVersions}});
    }

    session.commitTransaction();
    print("更新成功，事务已提交");
} catch (e) {
    session.abortTransaction();
    print("更新失败，事务已回滚，错误信息：" + e);
    throw e;
} finally {
    session.endSession();
}

注意事项

• 若集合数据量超过10万，建议分批处理，每批处理1万条数据开启一个事务，避免超过MongoDB默认60秒的事务生命周期限制。
• 执行前建议先对集合做全量备份，在测试环境验证逻辑无误后再线上执行。

方案2：单节点逐文档原子更新方案 #

若无法使用事务，可使用乐观锁实现单文档原子更新，每个文档的更新是原子操作，不会出现中间状态，通过比对更新前的文档快照保证没有其他写入干扰。

执行脚本

// 替换为实际的数据库名、集合名
const DB_NAME = "your_database";
const COLL_NAME = "your_collection";
const DELTA = NumberLong("9223372036854775808");
// 每批次处理文档数，可根据实际情况调整
const BATCH_SIZE = 1000;

const coll = db.getSiblingDB(DB_NAME).getCollection(COLL_NAME);
let totalUpdated = 0;

while (true) {
    const docs = coll.find({"versions.revision": {$lt: 0}}).limit(BATCH_SIZE).toArray();
    if (docs.length === 0) break;

    docs.forEach(doc => {
        let retry = 3;
        while (retry > 0) {
            const newVersions = doc.versions.map(elem => {
                if (elem.revision.lt(0)) {
                    return Object.assign({}, elem, {revision: elem.revision.add(DELTA)});
                }
                return elem;
            });
            // 乐观锁：更新条件带上原始versions快照，匹配不到说明文档被修改，重试
            const updateRes = coll.updateOne(
                {_id: doc._id, versions: doc.versions},
                {$set: {versions: newVersions}}
            );
            if (updateRes.matchedCount === 1) {
                totalUpdated++;
                break;
            }
            retry--;
            // 重新拉取最新的文档快照
            doc = coll.findOne({_id: doc._id});
        }
        if (retry === 0) {
            throw new Error(`文档_id ${doc._id} 更新重试失败，请手动检查`);
        }
    });
}

print(`更新完成，共更新 ${totalUpdated} 个文档`);

内容的提问来源于stack exchange，提问作者John Allison