Phi Accrual 故障检测器#

public class PhiAccrualFailureDetector {

    // 滑动窗口大小
    private final int windowSize;
    // 最小标准差（防止网络波动时 phi 值为负无穷）
    private final double minStdDeviation;
    // phi 值的阈值，超过这个值认为节点已死亡
    private final double threshold;
    // 心跳周期（用于计算初始 phi）
    private final Duration heartbeatInterval;

    // 历史心跳间隔的滑动窗口
    private final List<Long> history;
    // 最后一次收到心跳的时间
    private volatile long lastHeartbeat;

    public PhiAccrualFailureDetector(
            int windowSize,
            double threshold,
            Duration heartbeatInterval) {
        this.windowSize = windowSize;
        this.threshold = threshold;
        this.heartbeatInterval = heartbeatInterval;
        this.minStdDeviation = heartbeatInterval.toMillis() * 0.1; // 最小为周期的 10%
        this.history = new ArrayList<>();
        this.lastHeartbeat = System.currentTimeMillis();
    }

    /**
     * 记录一次心跳
     */
    public void heartbeat(String nodeId) {
        long now = System.currentTimeMillis();
        long interval = now - lastHeartbeat;

        synchronized (history) {
            history.add(interval);
            if (history.size() > windowSize) {
                history.remove(0);
            }
        }

        lastHeartbeat = now;
    }

    /**
     * 计算当前节点的 phi 值
     */
    public double phi(String nodeId) {
        long now = System.currentTimeMillis();
        long timeSinceLastHeartbeat = now - lastHeartbeat;

        synchronized (history) {
            if (history.isEmpty()) {
                // 没有历史数据，使用心跳周期作为均值
                double mean = heartbeatInterval.toMillis();
                return phi(heartbeatInterval.toMillis(), timeSinceLastHeartbeat, mean, minStdDeviation);
            }

            // 计算均值和标准差
            double mean = history.stream().mapToLong(Long::longValue).average().orElse(heartbeatInterval.toMillis());
            double variance = history.stream()
                    .mapToDouble(v -> Math.pow(v - mean, 2))
                    .average()
                    .orElse(minStdDeviation * minStdDeviation);
            double stdDeviation = Math.max(Math.sqrt(variance), minStdDeviation);

            return phi(mean, timeSinceLastHeartbeat, mean, stdDeviation);
        }
    }

    /**
     * 基于正态分布计算 phi
     */
    private double phi(double mean, double timeSinceLastHeartbeat,
                       double calculatedMean, double stdDeviation) {
        if (stdDeviation == 0) {
            // 标准差为 0 时，退化为固定超时
            return timeSinceLastHeartbeat > mean ? 1.0 : 0.0;
        }

        // 计算 Phi = -log10(1 - CDF(Δ))
        // CDF 是正态分布的累计分布函数
        double y = (timeSinceLastHeartbeat - calculatedMean) / stdDeviation;
        double probability = 1.0 - normalCDF(y);

        if (probability < 0.0000000000000001) {
            return 16.0; // 防止 log(0)
        }

        return -Math.log10(probability);
    }

    /**
     * 标准正态分布的累计分布函数（CDF）
     */
    private double normalCDF(double x) {
        // 使用误差函数逼近标准正态分布的 CDF
        return 0.5 * (1.0 + erf(x / Math.sqrt(2)));
    }

    /**
     * 误差函数逼近（Abramowitz and Stegun）
     */
    private double erf(double x) {
        double t = 1.0 / (1.0 + 0.5 * Math.abs(x));
        double tau = t * Math.exp(
            -x * x - 1.26551223
            + t * (1.00002368
            + t * (0.37409196
            + t * (0.09678418
            + t * (-0.18628806
            + t * (0.27886807
            + t * (-1.13520398
            + t * (1.48851587
            + t * (-0.82215223
            + t * 0.17087277))))))))));

        return x >= 0 ? 1.0 - tau : tau - 1.0;
    }

    /**
     * 判断节点是否已经死亡
     */
    public boolean isAvailable(String nodeId) {
        return phi(nodeId) < threshold;
    }

    /**
     * 获取可读的存活状态描述
     */
    public AvailabilityStatus getStatus(String nodeId) {
        double phi = phi(nodeId);
        if (phi < threshold) {
            return AvailabilityStatus.ALIVE;
        } else if (phi < threshold * 1.5) {
            return AvailabilityStatus.SUSPECT;
        } else {
            return AvailabilityStatus.DEAD;
        }
    }
}

enum AvailabilityStatus {
    ALIVE,      // 正常运行
    SUSPECT,    // 可疑状态
    DEAD        // 判定死亡
}

public class FailureDetectorDemo {

    public static void main(String[] args) throws InterruptedException {
        // 创建检测器：窗口大小 100，phi 阈值 8，心跳周期 1 秒
        PhiAccrualFailureDetector detector = new PhiAccrualFailureDetector(
            100,           // 窗口大小
            8.0,           // phi 阈值
            Duration.ofSeconds(1)
        );

        // 模拟节点注册
        String nodeId = "node-1";

        // 模拟正常心跳
        for (int i = 0; i < 10; i++) {
            detector.heartbeat(nodeId);
            System.out.printf("[正常] phi = %.2f, 可用 = %s%n",
                detector.phi(nodeId),
                detector.isAvailable(nodeId));
            Thread.sleep(1000);
        }

        // 模拟网络波动：心跳间隔变长
        System.out.println("\n[网络波动] 心跳间隔变长...");
        Thread.sleep(3000); // 3 秒没收到心跳
        System.out.printf("[网络波动] phi = %.2f, 可用 = %s%n",
            detector.phi(nodeId),
            detector.isAvailable(nodeId));

        // 模拟故障：长时间无心跳
        System.out.println("\n[故障模拟] 继续等待...");
        Thread.sleep(5000); // 再等 5 秒
        System.out.printf("[故障模拟] phi = %.2f, 可用 = %s%n",
            detector.phi(nodeId),
            detector.isAvailable(nodeId));
    }
}