Quorum 读写机制#

public class QuorumWrite {
    // N=3, W=2, R=2
    private int N = 3;
    private int W = 2;

    public boolean write(String key, String value, VectorClock vc) {
        // 1. 找到 key 所属的 N 个副本节点
        List<Node> replicas = locator.getNodes(key, N);

        // 2. 并行写入所有副本
        List<Future<Boolean>> futures = new ArrayList<>();
        for (Node node : replicas) {
            futures.add(executor.submit(() -> {
                return node.put(key, value, vc);
            }));
        }

        // 3. 等待 W 个节点确认
        int successCount = 0;
        List<Exception> failures = new ArrayList<>();

        for (Future<Boolean> f : futures) {
            try {
                if (f.get(writeTimeout, TimeUnit.MILLISECONDS)) {
                    successCount++;
                }
            } catch (Exception e) {
                failures.add(e);
            }
        }

        // 4. 判断是否达到 Quorum
        if (successCount >= W) {
            return true; // 写入成功
        } else {
            // 写入失败，但数据可能已经在某些节点上了
            // 需要触发后续修复（Read Repair 或 Merkle Tree）
            triggerAsyncRepair(key, value, failures);
            return false;
        }
    }
}

public class QuorumRead {
    private int N = 3;
    private int R = 2;

    public ReadResult read(String key) {
        // 1. 找到 key 所属的 N 个副本节点
        List<Node> replicas = locator.getNodes(key, N);

        // 2. 并行读取 R 个节点
        List<Future<Version>> futures = new ArrayList<>();
        for (int i = 0; i < R; i++) {
            Node node = replicas.get(i);
            futures.add(executor.submit(() -> {
                return node.get(key);
            }));
        }

        // 3. 收集响应并解决冲突
        List<Version> versions = new ArrayList<>();
        for (Future<Version> f : futures) {
            try {
                versions.add(f.get(readTimeout, TimeUnit.MILLISECONDS));
            } catch (Exception e) {
                // 节点超时，继续收集其他响应
            }
        }

        if (versions.isEmpty()) {
            throw new ReadFailureException("无法读取到足够副本");
        }

        // 4. 冲突解决：取最新版本
        Version latest = resolveConflict(versions);
        return new ReadResult(latest.value, latest.vc);
    }
}

public class ConflictResolver {
    // 方式1：Last-Write-Wins（最常用）
    public Version resolveConflict(List<Version> versions) {
        return versions.stream()
            .max(Comparator.comparing(v -> v.timestamp))
            .orElse(versions.get(0));
    }

    // 方式2：向量时钟比较（保持因果）
    public Version resolveConflictByVC(List<Version> versions) {
        for (int i = 0; i < versions.size(); i++) {
            for (int j = i + 1; j < versions.size(); j++) {
                Relation r = compareVC(versions.get(i).vc, versions.get(j).vc);
                if (r == Relation.BEFORE) {
                    return versions.get(j);
                }
                if (r == Relation.AFTER) {
                    return versions.get(i);
                }
                // CONCURRENT：需要更高层次的冲突解决
            }
        }
        // 并发：使用时间戳作为最终裁决
        return versions.stream()
            .max(Comparator.comparing(v -> v.timestamp))
            .orElse(versions.get(0));
    }

    // 方式3：应用层合并（VectorSet / CRDT）
    public Set<String> mergeSet(List<Set<String>> versions) {
        Set<String> result = new HashSet<>();
        for (Set<String> set : versions) {
            result.addAll(set);
        }
        return result; // 并集合并（购物车场景适用）
    }
}

public class ReadRepair {
    public void readWithRepair(String key) {
        // 读取 R 个节点
        List<Version> versions = quorumRead(key);

        // 取最新版本
        Version latest = resolveConflict(versions);

        // 异步修复其他节点（不阻塞读请求）
        for (Node node : replicas) {
            if (!node.hasLatest(key, latest)) {
                asyncRepair(node, key, latest);
            }
        }

        return latest.value;
    }
}

// 场景1：金融账户（强一致优先）
config.n = 5;
config.w = 4;  // 80% 确认才返回
config.r = 5;  // 全部节点读
// R+W=9 > 5 → 强一致，但慢

// 场景2：社交 Feed（写入快优先）
config.n = 3;
config.w = 1;  // 写入只要1个节点确认
config.r = 2;  // 读取至少2个
// R+W=3 = N → 最终一致，但写入极快

// 场景3：库存扣减（读取快优先）
config.n = 3;
config.w = 2;  // 写入等2个
config.r = 1;  // 读取只读1个（用版本号校验）
// 需要额外的乐观锁机制（版本号检测冲突）

// ❌ 错误：超时设置过短
Future<Boolean> future = node.putAsync(key, value);
if (future.get(100, TimeUnit.MILLISECONDS)) { // 100ms 太短
    return true;
}
// 节点其实正在写入，只是网络慢了一点
// 标记为失败，但数据已经在 2 个节点上了 → 数据不一致

// ✅ 正确：超时要合理
int nodeCount = N;
long baseTimeout = 10; // 每个节点基础超时
long totalTimeout = baseTimeout * nodeCount; // 考虑并发
// 同时加上重试机制