向量时钟与版本向量#

public class VectorClock {
    Map<String, Integer> clock = new HashMap<>();

    public void localEvent(String nodeId) {
        int current = clock.getOrDefault(nodeId, 0);
        clock.put(nodeId, current + 1);
    }
}

public void onMessageReceived(Message msg, String myNodeId) {
    Map<String, Integer> receivedVC = msg.getVectorClock();

    // 合并：每个分量取 max
    for (String nodeId : allNodes) {
        int local = clock.getOrDefault(nodeId, 0);
        int remote = receivedVC.getOrDefault(nodeId, 0);
        clock.put(nodeId, Math.max(local, remote));
    }

    // 本地时间戳 +1
    clock.put(myNodeId, clock.get(myNodeId) + 1);
}

public enum CausalityRelation {
    BEFORE,      // VC1 在 VC2 之前（VC1 是因）
    AFTER,       // VC1 在 VC2 之后（VC2 是因）
    EQUAL,       // 完全相同
    CONCURRENT,  // 并发（无因果关系）
}

public static CausalityRelation compare(
        Map<String, Integer> vc1,
        Map<String, Integer> vc2) {

    Set<String> allNodes = new HashSet<>(vc1.keySet());
    allNodes.addAll(vc2.keySet());

    boolean vc1AllLessOrEqual = true;
    boolean vc2AllLessOrEqual = true;

    for (String node : allNodes) {
        int v1 = vc1.getOrDefault(node, 0);
        int v2 = vc2.getOrDefault(node, 0);

        if (v1 < v2) vc1AllLessOrEqual = false;
        if (v2 < v1) vc2AllLessOrEqual = false;
    }

    if (vc1AllLessOrEqual && vc2AllLessOrEqual) return CausalityRelation.EQUAL;
    if (vc1AllLessOrEqual) return CausalityRelation.BEFORE;
    if (vc2AllLessOrEqual) return CausalityRelation.AFTER;
    return CausalityRelation.CONCURRENT;
}

// Cassandra 冲突解决：Last-Write-Wins（最常见）
public Version resolveConflict(Version v1, Version v2) {
    // 时间戳大的胜出
    if (v1.timestamp > v2.timestamp) {
        return v1;
    }
    return v2;
}

public Object resolveByLWW(List<VersionedValue> versions) {
    return versions.stream()
        .max(Comparator.comparingLong(v -> v.timestamp))
        .orElseThrow();
}

public List<VersionedValue> mergeByCausality(List<VersionedValue> versions) {
    List<VersionedValue> result = new ArrayList<>();

    for (int i = 0; i < versions.size(); i++) {
        boolean dominated = false;
        for (int j = 0; j < versions.size(); j++) {
            if (i == j) continue;

            CausalityRelation r = compare(versions.get(i).vc, versions.get(j).vc);
            if (r == CausalityRelation.BEFORE) {
                dominated = true; // 被其他版本因果覆盖，丢弃
                break;
            }
        }
        if (!dominated) {
            result.add(versions.get(i));
        }
    }
    return result; // 只保留非被覆盖的版本
}

// 读修复（Read Repair）：读取时主动修复
public class ReadRepair {
    public Version readWithRepair(String key) {
        // 1. 从 N 个副本读取
        List<Version> versions = quorumRead(key);

        // 2. 解决冲突
        Version latest = resolveConflict(versions);

        // 3. 异步修复过期副本（不阻塞读）
        for (Node node : replicas) {
            if (!node.has(latest)) {
                asyncRepair(node, key, latest);
            }
        }

        return latest;
    }
}

// 反熵（Anti-Entropy）：后台定期校验
public class AntiEntropy {
    // 使用 Merkle Tree 快速检测副本间差异
    // 差异部分进行增量同步

    public void verify(Node node1, Node node2) {
        MerkleTree tree1 = node1.buildMerkleTree();
        MerkleTree tree2 = node2.buildMerkleTree();

        // 比较 Merkle Tree 的哈希值
        // 不同子树代表需要同步的数据范围
    }
}

public class BoundedVectorClock {
    private static final int MAX_ENTRIES = 10;
    private Map<String, Integer> vc = new LinkedHashMap<>(MAX_ENTRIES);

    // 按访问顺序保留最近 MAX_ENTRIES 个节点
    public void put(String nodeId, int value) {
        if (vc.size() >= MAX_ENTRIES && !vc.containsKey(nodeId)) {
            // 删除最旧的条目
            String oldest = vc.keySet().iterator().next();
            vc.remove(oldest);
        }
        vc.put(nodeId, value);
    }
}

// ❌ 错误：完全依赖物理时钟
// 问题：NTP 延迟、时钟漂移导致时间戳失真

// ✅ 正确：使用混合逻辑时钟（HLC）
// Hybrid Logical Clock = 物理时间 + 逻辑计数器
public class HybridLogicalClock {
    long physical;   // 物理时间（来自本地时钟）
    int logical;      // 逻辑计数器

    public HybridLogicalClock receive(HybridLogicalClock remote) {
        this.physical = Math.max(this.physical, remote.physical) + 1;
        this.logical = (this.physical == remote.physical)
            ? Math.max(this.logical, remote.logical) + 1
            : 0;
        return this;
    }
}

// ❌ 错误：无条件截断版本向量
// 问题：截断后，无法判断某些版本的因果关系

// ✅ 正确：基于阈值的有界截断
public class VersionVector {
    private Map<String, Long> vv = new HashMap<>();
    private static final long MAX_VERSION = 1000;

    public void increment(String replicaId) {
        long current = vv.getOrDefault(replicaId, 0L) + 1;
        vv.put(replicaId, current);

        // 只在超过阈值时才截断
        if (vv.size() > MAX_VERSION && current > MAX_VERSION) {
            // 保留摘要，丢弃详细历史
            this.summary = computeSummary(vv);
            this.vv.clear();
            this.vv.put(replicaId, current);
        }
    }
}