Cassandra 中的 Gossip 实践#

public class Gossiper {

    // 内部状态
    private final Map<InetAddressAndPort, EndpointState> endpointStateMap;
    private final Map<InetAddressAndPort, Long> recentGossipHistory;

    /**
     * 模拟 Cassandra Gossip 的核心逻辑
     */
    public void Gossiper(ApplicationState state) {
        this.endpointStateMap = new ConcurrentHashMap<>();
        this.recentGossipHistory = new ConcurrentHashMap<>();
    }

    /**
     * 节点注册（节点加入时调用）
     */
    public void register(InetAddressAndPort endpoint) {
        EndpointState state = new EndpointState();
        state.setApplicationState(ApplicationState.STATUS,
            Value.serializer(new EndpointState.Change(Change.NEW, null)));
        state.setHeartBeatState(new HeartBeatState(0, getGeneration(endpoint)));

        endpointStateMap.put(endpoint, state);

        // 广播 Alive 消息
        disseminate(state);
    }

    /**
     * 更新应用状态
     */
    public void setApplicationState(InetAddressAndPort endpoint,
                                     ApplicationState key,
                                     Value value) {
        EndpointState state = endpointStateMap.computeIfAbsent(endpoint,
            k -> new EndpointState());

        state.setApplicationState(key, value);
        state.getHeartBeatState().incrementHeartBeat();

        // 通过 Gossip 传播
        disseminate(state);
    }

    /**
     * Gossip 传播入口
     */
    private void disseminate(EndpointState state) {
        // 随机选择 1-3 个节点进行 Gossip
        List<InetAddressAndPort> targets = selectRandomGossipTargets(3);

        for (InetAddressAndPort target : targets) {
            GossipDigestSyn synMessage = buildSynMessage(state);
            sendGossip(synMessage, target);
        }
    }

    /**
     * 处理 Gossip 响应
     */
    public void handleGossipDigestAck(GossipDigestAck ack) {
        for (GossipDigest delta : ack.getDelta()) {
            EndpointState remoteState = delta.getEndpointState();

            // 合并状态（以心跳计数较大的为准）
            endpointStateMap.compute(delta.getEndpoint(),
                (k, localState) -> mergeState(localState, remoteState));
        }
    }

    /**
     * 状态合并：心跳计数大的版本优先
     */
    private EndpointState mergeState(EndpointState local, EndpointState remote) {
        if (local == null) {
            return remote;
        }

        if (remote.getHeartBeatState().getGeneration()
                > local.getHeartBeatState().getGeneration()) {
            return remote;
        }

        if (remote.getHeartBeatState().getHeartBeatNumber()
                > local.getHeartBeatState().getHeartBeatNumber()) {
            // 合并 ApplicationState
            for (Map.Entry<ApplicationState, Value> entry :
                    remote.getApplicationStateMap().entrySet()) {
                local.setApplicationState(entry.getKey(), entry.getValue());
            }
            return local;
        }

        return local;
    }
}

public class FailureDetector {

    private final double phiThreshold;
    private final Map<InetAddressAndPort, PhiAccrualState> stateMap;

    public FailureDetector(double phiThreshold) {
        this.phiThreshold = phiThreshold;
        this.stateMap = new ConcurrentHashMap<>();
    }

    /**
     * 更新心跳（当收到节点的 Gossip 时调用）
     */
    public void record(InetAddressAndPort endpoint, long arrivalTime) {
        PhiAccrualState state = stateMap.computeIfAbsent(endpoint,
            k -> new PhiAccrualState());
        state.addSample(arrivalTime);
    }

    /**
     * 计算某个节点的 phi 值
     */
    public double getPhi(InetAddressAndPort endpoint) {
        PhiAccrualState state = stateMap.get(endpoint);
        if (state == null) {
            return 0.0;
        }

        long now = System.currentTimeMillis();
        long timeSinceLastHeartbeat = now - state.lastHeartbeat();

        // Cassandra 的简化版 phi 计算
        // 假设心跳间隔服从指数分布
        double mean = state.meanInterval();
        double phi = (timeSinceLastHeartbeat - mean) / mean;

        return phi;
    }

    /**
     * 判断节点是否可疑（phi > 阈值）
     */
    public boolean isSuspect(InetAddressAndPort endpoint) {
        return getPhi(endpoint) >= phiThreshold;
    }
}

public class DatacenterGossipRateLimiter {

    private final double crossDcRate;
    private final Random random = new Random();

    public DatacenterGossipRateLimiter(double crossDcRate) {
        this.crossDcRate = crossDcRate; // 默认 0.33
    }

    /**
     * 判断某次 Gossip 是否允许跨数据中心
     */
    public boolean allowCrossDatacenter(InetAddressAndPort target,
                                         String myDc,
                                         String targetDc) {
        // 同 DC 总是允许
        if (myDc.equals(targetDc)) {
            return true;
        }

        // 跨 DC 按比例允许
        return random.nextDouble() < crossDcRate;
    }
}

// 关键 Gossip 监控指标
public interface GossipMetrics {
    // 发送/接收的消息数
    MBeanAttributeInfo "Gossiper-StartedAt"
    MBeanAttributeInfo "SimpleMetrics-AppliedMutations"

    // Endpoint 状态
    MBeanAttributeInfo "EndpointState-UpSince"
    MBeanAttributeInfo "EndpointState-Alive"
    MBeanAttributeInfo "EndpointState-Versions"
}