因果一致性#

// 向量时钟的三条核心规则

// 规则1：本地操作后，时钟递增
public class Node {
    Map<String, Integer> vectorClock = new HashMap<>();

    public void localOperation() {
        int current = vectorClock.getOrDefault(nodeId, 0);
        vectorClock.put(nodeId, current + 1);
    }
}

// 规则2：发送消息时，附上当前向量时钟
public void send(Message msg) {
    Message withClock = new Message(msg);
    withClock.setVectorClock(vectorClock.copy()); // 复制当前时钟
    network.send(withClock);
}

// 规则3：收到消息时，合并时钟（取最大值）
public void receive(Message msg) {
    Map<String, Integer> received = msg.getVectorClock();

    // 合并：每个节点取 max
    for (String nodeId : allNodes) {
        int local = vectorClock.getOrDefault(nodeId, 0);
        int remote = received.getOrDefault(nodeId, 0);
        vectorClock.put(nodeId, Math.max(local, remote));
    }

    // 最后本地时钟 +1
    vectorClock.put(this.nodeId, vectorClock.get(this.nodeId) + 1);
}

// 判断两个向量时钟的因果关系

// 定义：VC1 < VC2 表示 VC1 在 VC2 之前（VC1 是因）
public static Relation compare(Map<String, Integer> vc1, Map<String, Integer> vc2) {
    boolean allLessOrEqual = true;
    boolean allEqual = true;

    // 合并所有节点
    Set<String> allNodes = new HashSet<>(vc1.keySet());
    allNodes.addAll(vc2.keySet());

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

        if (v1 < v2) allLessOrEqual = false;
        if (v1 != v2) allEqual = false;
    }

    if (allEqual) return Relation.EQUAL;
    if (allLessOrEqual) return Relation.BEFORE; // vc1 在 vc2 之前
    // vc2 在 vc1 之前
    if (isAllGreaterOrEqual(vc2, vc1)) return Relation.AFTER;

    // 否则：并发（无因果关系）
    return Relation.CONCURRENT;
}

// COPS 的因果一致性实现框架
public class COPSNode {
    private Map<String, Object> localStore = new HashMap<>();
    private Map<String, VectorClock> metadata = new HashMap<>();

    // 本地写入
    public void put(String key, String value) {
        VectorClock vc = incrementLocalClock();

        localStore.put(key, value);
        metadata.put(key, vc);

        // 异步复制到其他节点
        replicateAsync(key, value, vc);
    }

    // 因果一致读取
    public String get(String key) {
        // 等待依赖的所有操作都应用
        VectorClock dependsOn = metadata.get(key);
        waitForDependencies(dependsOn);

        return localStore.get(key);
    }
}

// 微博评论的因果一致性实现
public class CommentService {
    // 评论的因果依赖链
    public class Comment {
        String weiboId;        // 依赖微博
        String parentId;       // 依赖父评论（回复场景）
        VectorClock vc;        // 向量时钟
    }

    // 发布评论
    public void postComment(Comment comment) {
        // 1. 确认依赖存在
        if (comment.parentId != null) {
            Comment parent = commentStore.get(comment.parentId);
            waitForVC(parent.vc); // 等待父评论的因果依赖
        }

        // 2. 记录自己的向量时钟
        comment.vc = incrementLocalClock();

        // 3. 存储并广播
        commentStore.put(comment.id, comment);
        broadcast(comment);
    }

    // 读取评论（因果一致）
    public List<Comment> getComments(String weiboId) {
        List<Comment> result = new ArrayList<>();

        // 1. 获取微博的因果依赖
        Weibo weibo = weiboStore.get(weiboId);
        waitForVC(weibo.vc); // 确保微博已加载

        // 2. 只加载因果相关的评论
        for (Comment c : commentStore.getByWeibo(weiboId)) {
            if (isCausallyReady(c)) { // 确保因果依赖已满足
                result.add(c);
            }
        }

        return result;
    }
}

// 协作编辑的因果一致性
public class CollaborativeEditor {
    private Map<String, Operation> ops = new ConcurrentHashMap<>();

    public void apply(Operation op) {
        // 1. 检查因果依赖
        for (String depId : op.getDependencies()) {
            if (!ops.containsKey(depId)) {
                waitFor(ops, depId); // 等待依赖操作到达
            }
        }

        // 2. 确认依赖后应用操作
        VectorClock newVC = mergeVC(op.getVC(), localVC);
        op.setVC(newVC);
        ops.put(op.getId(), op);

        // 3. 应用到文档
        applyToDocument(op);
    }
}

// ❌ 错误：向量时钟随节点数线性增长
// 1000个节点 → 每个消息携带1000维向量
// 1000万消息 → 内存爆炸

// ✅ 正确：使用有界向量时钟
public class BoundedVectorClock {
    // 只记录最近 N 个节点的时钟，或使用截断策略
    private static final int MAX_SIZE = 10;
    private Map<String, Integer> vc = new LinkedHashMap<>(MAX_SIZE);

    public void put(String nodeId, int value) {
        if (vc.size() >= MAX_SIZE) {
            // 删除最旧的条目
            String oldest = vc.keySet().iterator().next();
            vc.remove(oldest);
        }
        vc.put(nodeId, value);
    }
}

// ❌ 错误：认为因果一致就是最终一致
// 最终一致：停止写入后最终收敛，但不保证顺序
// 因果一致：收敛时保证因果顺序

// ✅ 正确：因果一致是最终一致的子集
// 因果一致 → 一定最终一致（收敛时有因果保证）
// 最终一致 ↛ 因果一致（收敛时可能因果乱序）