分布式系统故障模型#

// 崩溃故障检测：心跳机制
public class CrashDetector {
    private final Map<String, Long> lastHeartbeat = new ConcurrentHashMap<>();
    private final long timeoutMs = 5000;

    public void onHeartbeat(String nodeId) {
        lastHeartbeat.put(nodeId, System.currentTimeMillis());
    }

    public Set<String> detectCrashedNodes() {
        long now = System.currentTimeMillis();
        Set<String> crashed = new HashSet<>();

        for (Map.Entry<String, Long> entry : lastHeartbeat.entrySet()) {
            if (now - entry.getValue() > timeoutMs) {
                crashed.add(entry.getKey());
            }
        }
        return crashed;
    }
}

// 遗漏故障处理：可靠消息通道
public class ReliableChannel {
    private final int maxRetries = 3;
    private final long retryDelayMs = 100;

    public boolean sendWithRetry(String nodeId, Message msg) {
        for (int attempt = 0; attempt < maxRetries; attempt++) {
            try {
                network.send(nodeId, msg);
                if (waitForAck(nodeId, msg.getId(), 1000)) {
                    return true; // 消息被确认
                }
            } catch (NetworkException e) {
                // 消息遗漏，等待后重试
                sleep(retryDelayMs);
            }
        }
        return false; // 重试耗尽
    }
}

// 时序故障处理：自适应超时
public class AdaptiveTimeout {
    // 基于历史数据动态调整超时
    private static final double RTT_ALPHA = 0.3;
    private static final double RTTVAR_ALPHA = 0.3;

    private double estimatedRTT = 1000; // 估计往返时间
    private double rttvar = 0;          // RTT 方差

    public void recordRTT(long actualRTT) {
        rttvar = RTTVAR_ALPHA * rttvar
               + RTT_ALPHA * Math.abs(actualRTT - estimatedRTT);
        estimatedRTT = RTT_ALPHA * estimatedRTT
                      + (1 - RTT_ALPHA) * actualRTT;
    }

    public long getTimeout() {
        // TCP 标准：timeout = estimatedRTT + 4 * rttvar
        return (long)(estimatedRTT + 4 * rttvar);
    }
}

// 响应故障处理：验证机制
public class ResponseVerifier {
    // 简单验证：校验和
    public boolean verifyChecksum(Message msg) {
        int expected = msg.getChecksum();
        int actual = computeChecksum(msg.getData());
        return expected == actual;
    }

    // 复杂验证：拜占庭协议（后面会讲）
    public boolean verifyByzantine(List<ReadResponse> responses, int f) {
        // 如果 f 个节点返回错误答案，
        // 2f+1 个节点中正确答案是多数派
        return responses.stream()
            .filter(this::isCorrectResponse)
            .count() > f;
    }
}

// 拜占庭将军问题：口信消息型
// n 个将军，其中 m 个是叛徒（拜占庭节点）
// 算法要求：n >= 3m + 1

public class ByzantineConsensus {
    private int f; // 可容忍的拜占庭节点数
    private int n; // 总节点数

    public ByzantineConsensus(int n, int f) {
        if (n < 3 * f + 1) {
            throw new IllegalArgumentException(
                "Need n >= 3f+1 for Byzantine fault tolerance");
        }
        this.n = n;
        this.f = f;
    }

    // PBFT（Practical Byzantine Fault Tolerance）的简化逻辑
    public boolean reachConsensus(String value) {
        // 1. Pre-prepare：主节点广播提议
        // 2. Prepare：所有节点互相通信，验证提议
        //    如果收到 >= 2f 个 Prepare 消息，节点进入 Prepared 状态
        // 3. Commit：进入 Prepared 状态的节点广播 Commit
        //    如果收到 >= 2f+1 个 Commit 消息，提交决议

        int prepareCount = collectPrepareMessages();
        int commitCount = collectCommitMessages();

        return prepareCount >= 2 * f + 1
            && commitCount >= 2 * f + 1;
    }
}

public class HeartbeatFailureDetector {
    private final ScheduledExecutorService scheduler =
        Executors.newScheduledThreadPool(1);
    private final Map<String, AtomicLong> lastHeartbeat = new ConcurrentHashMap<>();
    private final long phi = 8; // 检测敏感度

    public void start(Node node) {
        // 发送心跳
        scheduler.scheduleAtFixedRate(() -> {
            network.broadcast(new Heartbeat(localNodeId));
        }, 0, 100, TimeUnit.MILLISECONDS);

        // 处理心跳响应
        scheduler.scheduleAtFixedRate(() -> {
            for (String peer : peers) {
                long delay = computeDelay(peer);
                long phiValue = calculatePhi(delay);
                if (phiValue > phi) {
                    markSuspected(peer);
                }
            }
        }, 0, 100, TimeUnit.MILLISECONDS);
    }

    // Phi Accrual Detector（Swim 使用的）
    private double calculatePhi(long currentDelay) {
        double mean = history.getMean();
        double variance = history.getVariance();
        double sd = Math.sqrt(variance);

        // Φ = -log10(1 - CDF(delay))
        return -Math.log10(1 - normalCDF(currentDelay, mean, sd)) / Math.log10(10);
    }
}

public class GossipFailureDetector {
    private final Map<String, GossipState> state = new ConcurrentHashMap<>();

    // 每秒执行一次
    public void gossip() {
        // 1. 选择一个随机节点交换信息
        String target = randomPeer();
        Map<String, EndpointState> delta = collectLocalState();

        // 2. 交换并合并状态
        Map<String, EndpointState> remoteState = exchange(target, delta);
        mergeState(remoteState);

        // 3. 标记超时节点
        markOverdueNodes();
    }

    // 故障判定：超过阈值没更新
    public boolean isAlive(String nodeId) {
        GossipState gs = state.get(nodeId);
        if (gs == null) return false;

        long sinceLastUpdate = System.currentTimeMillis() - gs.getLastUpdateTime();
        return sinceLastUpdate < failureTimeoutMs;
    }
}

// 崩溃恢复后的日志同步
public class LogRecovery {
    public void recover(Node node) {
        // 1. 从 Leader 获取最新的 commitIndex
        long commitIndex = fetchCommitIndexFromLeader(node.leaderId);

        // 2. 补全本地缺失的日志
        List<LogEntry> missingEntries =
            fetchMissingEntries(node.id, node.lastLogIndex);

        for (LogEntry entry : missingEntries) {
            appendEntry(entry);
        }

        // 3. 应用到状态机
        applyToStateMachine(commitIndex);

        // 4. 恢复正常服务
        node.setState(NodeState.FOLLOWER);
    }
}