Akka集群多线程环境下Couchbase CAS失效，计数重复问题求助

问题根源 #

• 竞态窗口引发并发冲突：代码中先执行bucket.exists(documentId)检查文档存在性，这个操作和后续的getAndLock之间存在时间间隙。在Akka集群的分布式环境下，多个节点可能同时通过exists校验，进而同时进入更新逻辑，导致多个请求同时修改count值，出现重复计数。
• 锁与更新操作分离：你在getAndLock后先解锁再执行upsert，解锁后到upsert执行前的窗口内，其他节点可获取文档锁并修改数据，当前请求的upsert会覆盖他人修改，或与其他更新操作冲突，最终导致计数重复。
• CAS值使用错误：代码未正确获取getAndLock返回文档的CAS值，也未在upsert时用CAS保证更新原子性。无CAS校验的upsert会直接覆盖文档，无视期间的其他修改。

解决方案 #

最优方案：使用Couchbase原生原子计数API #

Couchbase提供了集群层面的原子递增方法counter，无需手动处理锁和CAS，直接实现安全计数：

// 原子递增count，文档不存在则初始化为1（第二个参数为步长，第三个为初始值）
long newCount = bucket.counter(documentId, 1, 1);
log.info("count : {}", newCount);

自定义实现修正方案 #

若需自行实现更新逻辑，需修正以下几点：

1. 删除独立的exists检查，直接尝试获取文档并处理不存在的场景；
2. 在锁内完成更新，使用CAS保证原子性，避免锁与更新分离；
3. 添加重试机制处理CAS更新失败的情况。

修正后的代码示例：

JsonDocument doc = bucket.getAndLock(documentId, 10); // 加锁10秒
if (doc != null) {
    try {
        int currentCount = doc.content().getInt("count");
        doc.content().put("count", currentCount + 1);
        // 用CAS值原子更新，确保只有当前锁的版本能被修改
        if (bucket.replace(doc)) {
            log.info("count : {}", doc.content().getInt("count"));
        } else {
            log.warn("Count update failed, retrying...");
            // 此处可添加重试逻辑，比如循环重试或用Akka的调度器重试
        }
    } finally {
        // 无论更新成功与否，最终解锁文档
        bucket.unlock(documentId, doc.cas());
    }
} else {
    // 文档不存在，创建初始count为1的文档
    JsonObject content = JsonObject.create().put("count", 1);
    JsonDocument newDoc = JsonDocument.create(documentId, content);
    bucket.upsert(newDoc);
    log.info("count : 1");
}

内容的提问来源于stack exchange，提问作者Jingon Park