【系统架构设计百科】Netflix 架构全景：混沌工程的诞生地

/**
 * Netflix 在迁移过程中使用的路由层伪代码
 * 通过功能开关决定请求走数据中心还是 AWS
 */
public class MigrationRouter {

    private final FeatureToggleService toggleService;
    private final DataCenterClient dcClient;
    private final AwsClient awsClient;

    public MigrationRouter(FeatureToggleService toggleService,
                           DataCenterClient dcClient,
                           AwsClient awsClient) {
        this.toggleService = toggleService;
        this.dcClient = dcClient;
        this.awsClient = awsClient;
    }

    public Response route(Request request) {
        String featureName = request.getFeatureName();

        if (toggleService.isAwsEnabled(featureName)) {
            try {
                Response awsResponse = awsClient.forward(request);
                // 迁移初期同时发送到数据中心做结果比对
                if (toggleService.isShadowMode(featureName)) {
                    Response dcResponse = dcClient.forward(request);
                    compareAndLog(awsResponse, dcResponse);
                }
                return awsResponse;
            } catch (Exception e) {
                // 降级回数据中心
                return dcClient.forward(request);
            }
        }

        return dcClient.forward(request);
    }

    private void compareAndLog(Response aws, Response dc) {
        if (!aws.equals(dc)) {
            MetricsLogger.logDivergence(aws, dc);
        }
    }
}

/**
 * Zuul 1.x 过滤器模型
 * 请求依次经过 Pre、Route、Post 三类过滤器
 */
public abstract class ZuulFilter implements Comparable<ZuulFilter> {

    /**
     * 过滤器类型：pre、route、post、error
     */
    public abstract String filterType();

    /**
     * 过滤器执行顺序，数字越小越先执行
     */
    public abstract int filterOrder();

    /**
     * 是否应该执行此过滤器
     */
    public abstract boolean shouldFilter();

    /**
     * 过滤器核心逻辑
     */
    public abstract Object run();
}

/**
 * 限流过滤器示例
 */
public class RateLimitFilter extends ZuulFilter {

    private final RateLimiter rateLimiter;

    public RateLimitFilter(int requestsPerSecond) {
        this.rateLimiter = RateLimiter.create(requestsPerSecond);
    }

    @Override
    public String filterType() {
        return "pre";
    }

    @Override
    public int filterOrder() {
        return 5;
    }

    @Override
    public boolean shouldFilter() {
        return true;
    }

    @Override
    public Object run() {
        RequestContext ctx = RequestContext.getCurrentContext();
        if (!rateLimiter.tryAcquire()) {
            ctx.setSendZuulResponse(false);
            ctx.setResponseStatusCode(429);
            ctx.setResponseBody("{\"error\": \"rate limit exceeded\"}");
        }
        return null;
    }
}

/**
 * Zuul 2.x 基于 Netty 的异步过滤器模型
 */
public abstract class HttpInboundSyncFilter extends HttpInboundFilter {

    @Override
    public Observable<HttpRequestMessage> applyAsync(
            HttpRequestMessage request) {
        return Observable.just(apply(request));
    }

    /**
     * 同步过滤器的简化接口
     */
    public abstract HttpRequestMessage apply(HttpRequestMessage request);
}

/**
 * 异步认证过滤器示例
 */
public class AsyncAuthFilter extends HttpInboundFilter {

    private final AuthService authService;

    public AsyncAuthFilter(AuthService authService) {
        this.authService = authService;
    }

    @Override
    public int filterOrder() {
        return 10;
    }

    @Override
    public boolean shouldFilter(HttpRequestMessage request) {
        return request.getHeaders().contains("Authorization");
    }

    @Override
    public Observable<HttpRequestMessage> applyAsync(
            HttpRequestMessage request) {
        String token = request.getHeaders()
                .getFirst("Authorization");

        return authService.validateTokenAsync(token)
                .map(authResult -> {
                    if (authResult.isValid()) {
                        request.getContext()
                                .put("userId", authResult.getUserId());
                    } else {
                        request.getContext()
                                .put("zuulResponse", createUnauthorized());
                    }
                    return request;
                });
    }
}

# Spring Cloud Gateway 配置示例
spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - StripPrefix=1
            - name: CircuitBreaker
              args:
                name: userServiceCB
                fallbackUri: forward:/fallback/users
            - name: RequestRateLimiter
              args:
                redis-rate-limiter.replenishRate: 100
                redis-rate-limiter.burstCapacity: 200
                key-resolver: "#{@userKeyResolver}"

        - id: video-service
          uri: lb://video-service
          predicates:
            - Path=/api/videos/**
            - Header=X-Request-Type, streaming
          filters:
            - StripPrefix=1
            - name: Retry
              args:
                retries: 3
                statuses: BAD_GATEWAY
                methods: GET
                backoff:
                  firstBackoff: 50ms
                  maxBackoff: 500ms
                  factor: 2

/**
 * EVCache 客户端使用示例
 * 展示基本的缓存读写操作和容错处理
 */
public class UserProfileCache {

    private final EVCache evCache;
    private final UserProfileDao userProfileDao;
    private static final int TTL_SECONDS = 3600;
    private static final String KEY_PREFIX = "uprof:";

    public UserProfileCache() {
        this.evCache = new EVCache.Builder()
                .setAppName("user-profile")
                .setCachePrefix(KEY_PREFIX)
                .enableRetries()
                .setDefaultTTL(TTL_SECONDS)
                .build();
        this.userProfileDao = new UserProfileDao();
    }

    /**
     * 获取用户资料，优先从缓存读取
     */
    public UserProfile getUserProfile(String userId) {
        String cacheKey = KEY_PREFIX + userId;

        try {
            // 尝试从本地区域缓存读取
            UserProfile cached = evCache.get(cacheKey, 
                    UserProfile.class);
            if (cached != null) {
                return cached;
            }
        } catch (EVCacheException e) {
            // 缓存读取失败不应阻塞业务
            Metrics.counter("evcache.read.error").increment();
        }

        // 缓存未命中，从数据库加载
        UserProfile profile = userProfileDao.findById(userId);
        if (profile != null) {
            try {
                // 写入缓存，异步复制到其他区域
                evCache.set(cacheKey, profile, TTL_SECONDS);
            } catch (EVCacheException e) {
                Metrics.counter("evcache.write.error").increment();
            }
        }
        return profile;
    }

    /**
     * 批量获取用户资料
     * EVCache 支持批量操作以减少网络往返
     */
    public Map<String, UserProfile> batchGetProfiles(
            List<String> userIds) {
        List<String> cacheKeys = userIds.stream()
                .map(id -> KEY_PREFIX + id)
                .collect(Collectors.toList());

        try {
            Map<String, UserProfile> cached = 
                    evCache.getBulk(cacheKeys, UserProfile.class);

            // 找出缓存未命中的用户
            List<String> missedIds = userIds.stream()
                    .filter(id -> !cached.containsKey(KEY_PREFIX + id))
                    .collect(Collectors.toList());

            if (!missedIds.isEmpty()) {
                Map<String, UserProfile> fromDb = 
                        userProfileDao.findByIds(missedIds);
                // 回填缓存
                fromDb.forEach((id, profile) -> {
                    try {
                        evCache.set(KEY_PREFIX + id, 
                                profile, TTL_SECONDS);
                    } catch (EVCacheException e) {
                        // 静默处理
                    }
                });
                cached.putAll(fromDb);
            }
            return cached;
        } catch (EVCacheException e) {
            // 缓存完全不可用时降级到数据库
            return userProfileDao.findByIds(userIds);
        }
    }
}

/**
 * EVCache 一致性哈希分片逻辑（简化版）
 */
public class EVCacheShardRouter {

    private final TreeMap<Long, String> hashRing;
    private final int virtualNodes;

    public EVCacheShardRouter(List<String> servers, 
                               int virtualNodes) {
        this.hashRing = new TreeMap<>();
        this.virtualNodes = virtualNodes;

        for (String server : servers) {
            for (int i = 0; i < virtualNodes; i++) {
                long hash = murmurHash(server + ":" + i);
                hashRing.put(hash, server);
            }
        }
    }

    /**
     * 根据缓存键找到对应的服务器节点
     */
    public String getServer(String key) {
        long hash = murmurHash(key);
        Map.Entry<Long, String> entry = hashRing.ceilingEntry(hash);
        if (entry == null) {
            entry = hashRing.firstEntry();
        }
        return entry.getValue();
    }

    /**
     * 获取副本节点列表（用于冗余读写）
     */
    public List<String> getReplicaServers(String key, 
                                           int replicaCount) {
        List<String> replicas = new ArrayList<>();
        long hash = murmurHash(key);
        Map.Entry<Long, String> entry = hashRing.ceilingEntry(hash);

        Iterator<Map.Entry<Long, String>> iterator;
        if (entry != null) {
            iterator = hashRing.tailMap(hash, true)
                    .entrySet().iterator();
        } else {
            iterator = hashRing.entrySet().iterator();
        }

        while (replicas.size() < replicaCount 
                && replicas.size() < hashRing.size()) {
            if (!iterator.hasNext()) {
                iterator = hashRing.entrySet().iterator();
            }
            String server = iterator.next().getValue();
            if (!replicas.contains(server)) {
                replicas.add(server);
            }
        }
        return replicas;
    }

    private long murmurHash(String key) {
        // MurmurHash3 实现省略
        return Math.abs(key.hashCode());
    }
}

"""
Chaos Monkey 核心逻辑（简化版）
随机选择一个运行中的实例并终止它
"""
import random
import logging
from datetime import datetime, time
from typing import List, Optional

logger = logging.getLogger("chaos_monkey")


class ChaosMonkey:
    """
    混沌猴子：在工作时间随机终止生产环境的实例
    """

    def __init__(self, cloud_client, config):
        self.cloud_client = cloud_client
        self.config = config
        self.termination_history = []

    def is_within_operating_hours(self) -> bool:
        """混沌猴子只在工作时间运行，确保有工程师值班"""
        now = datetime.now()
        start = time(9, 0)   # 上午 9 点
        end = time(15, 0)    # 下午 3 点
        return (now.weekday() < 5 
                and start <= now.time() <= end)

    def get_eligible_instances(self, 
                                app_name: str) -> List[dict]:
        """获取可被终止的实例列表"""
        instances = self.cloud_client.describe_instances(
            app_name=app_name,
            status="running"
        )
        # 排除受保护的实例
        return [
            inst for inst in instances
            if not inst.get("chaos_monkey_opt_out", False)
        ]

    def select_victim(self, 
                       instances: List[dict]) -> Optional[dict]:
        """随机选择一个"受害者"实例"""
        if not instances:
            return None

        # 基于分组策略选择
        strategy = self.config.get(
            "grouping_strategy", "app")

        if strategy == "app":
            return random.choice(instances)
        elif strategy == "stack":
            # 按部署栈分组，每组最多终止一个
            stacks = {}
            for inst in instances:
                stack = inst.get("stack", "default")
                stacks.setdefault(stack, []).append(inst)
            selected_stack = random.choice(
                list(stacks.values()))
            return random.choice(selected_stack)

        return random.choice(instances)

    def terminate_instance(self, instance: dict) -> bool:
        """终止选中的实例"""
        instance_id = instance["instance_id"]
        app_name = instance["app_name"]

        logger.info(
            "Chaos Monkey 正在终止实例: %s (应用: %s)",
            instance_id, app_name
        )

        try:
            self.cloud_client.terminate_instance(instance_id)
            self.termination_history.append({
                "instance_id": instance_id,
                "app_name": app_name,
                "timestamp": datetime.now().isoformat(),
                "region": instance.get("region", "unknown")
            })
            logger.info("实例 %s 已成功终止", instance_id)
            return True
        except Exception as e:
            logger.error(
                "终止实例 %s 失败: %s", instance_id, str(e))
            return False

    def run(self, app_names: List[str]) -> None:
        """执行一轮混沌实验"""
        if not self.is_within_operating_hours():
            logger.info("当前不在工作时间，跳过本轮执行")
            return

        for app_name in app_names:
            probability = self.config.get(
                "probability", 1.0)
            if random.random() > probability:
                logger.info(
                    "应用 %s 本轮未被选中", app_name)
                continue

            instances = self.get_eligible_instances(app_name)
            victim = self.select_victim(instances)
            if victim:
                self.terminate_instance(victim)

/**
 * 混沌工程实验框架（简化版）
 * 体现稳态假说和自动化验证的思想
 */
public class ChaosExperiment {

    private final String experimentName;
    private final SteadyStateHypothesis hypothesis;
    private final FaultInjector faultInjector;
    private final MetricsCollector metrics;

    public ChaosExperiment(String experimentName,
                           SteadyStateHypothesis hypothesis,
                           FaultInjector faultInjector,
                           MetricsCollector metrics) {
        this.experimentName = experimentName;
        this.hypothesis = hypothesis;
        this.faultInjector = faultInjector;
        this.metrics = metrics;
    }

    public ExperimentResult execute() {
        // 第一步：验证稳态假说成立（实验前基线）
        SteadyStateSnapshot baseline = 
                hypothesis.captureSnapshot();
        if (!hypothesis.isWithinBounds(baseline)) {
            return ExperimentResult.aborted(
                "系统未处于稳态，终止实验");
        }

        // 第二步：注入故障
        FaultHandle fault = null;
        try {
            fault = faultInjector.inject();

            // 第三步：观察系统行为
            Thread.sleep(
                faultInjector.getObservationPeriodMs());

            // 第四步：验证稳态假说是否仍然成立
            SteadyStateSnapshot afterFault = 
                    hypothesis.captureSnapshot();

            boolean hypothesisHolds = 
                    hypothesis.isWithinBounds(afterFault);

            return new ExperimentResult(
                experimentName,
                hypothesisHolds,
                baseline,
                afterFault,
                metrics.collectDuring(fault.getStartTime(), 
                        System.currentTimeMillis())
            );

        } catch (Exception e) {
            return ExperimentResult.failed(
                experimentName, e.getMessage());
        } finally {
            // 第五步：回滚故障注入
            if (fault != null) {
                fault.rollback();
            }
        }
    }
}

/**
 * 稳态假说定义示例：播放启动速率
 */
public class StreamStartsPerSecondHypothesis 
        implements SteadyStateHypothesis {

    private final double minSPS;
    private final double maxSPS;
    private final MetricsClient metricsClient;

    public StreamStartsPerSecondHypothesis(
            double minSPS, double maxSPS,
            MetricsClient metricsClient) {
        this.minSPS = minSPS;
        this.maxSPS = maxSPS;
        this.metricsClient = metricsClient;
    }

    @Override
    public SteadyStateSnapshot captureSnapshot() {
        double currentSPS = metricsClient.query(
            "stream_starts_per_second", 
            Duration.ofMinutes(5));
        return new SteadyStateSnapshot("SPS", currentSPS);
    }

    @Override
    public boolean isWithinBounds(
            SteadyStateSnapshot snapshot) {
        double value = snapshot.getValue();
        return value >= minSPS && value <= maxSPS;
    }
}

/**
 * Eureka 客户端注册与发现示例
 */
@EnableEurekaClient
@SpringBootApplication
public class VideoServiceApplication {

    public static void main(String[] args) {
        SpringApplication.run(
            VideoServiceApplication.class, args);
    }
}

/**
 * 使用 Eureka 进行服务发现和调用
 */
@Service
public class VideoMetadataService {

    private final EurekaClient eurekaClient;
    private final RestTemplate restTemplate;

    public VideoMetadataService(EurekaClient eurekaClient,
                                 RestTemplate restTemplate) {
        this.eurekaClient = eurekaClient;
        this.restTemplate = restTemplate;
    }

    public VideoMetadata getMetadata(String videoId) {
        // 从 Eureka 获取服务实例列表
        Application app = eurekaClient.getApplication(
            "metadata-service");
        List<InstanceInfo> instances = app.getInstances();

        // 选择一个健康的实例（简化版负载均衡）
        InstanceInfo instance = instances.stream()
                .filter(i -> i.getStatus() == 
                        InstanceInfo.InstanceStatus.UP)
                .findFirst()
                .orElseThrow(() -> new RuntimeException(
                    "无可用的 metadata-service 实例"));

        String url = String.format(
            "http://%s:%d/api/metadata/%s",
            instance.getHostName(),
            instance.getPort(),
            videoId
        );

        return restTemplate.getForObject(
            url, VideoMetadata.class);
    }
}

/**
 * Hystrix 熔断器使用示例
 */
public class RecommendationCommand 
        extends HystrixCommand<List<Video>> {

    private final String userId;
    private final RecommendationClient client;

    public RecommendationCommand(String userId,
                                  RecommendationClient client) {
        super(HystrixCommandGroupKey.Factory.asKey(
            "RecommendationService"));
        this.userId = userId;
        this.client = client;
    }

    @Override
    protected List<Video> run() throws Exception {
        // 正常调用推荐服务
        return client.getRecommendations(userId);
    }

    @Override
    protected List<Video> getFallback() {
        // 降级策略：返回热门视频列表
        return FallbackDataSource.getPopularVideos();
    }
}

/**
 * Hystrix 配置示例
 */
@Configuration
public class HystrixConfig {

    @PostConstruct
    public void configureHystrix() {
        ConfigurationManager.getConfigInstance()
            .setProperty(
                "hystrix.command.default"
                + ".circuitBreaker.requestVolumeThreshold",
                20)       // 20 个请求后开始统计
            ;
        ConfigurationManager.getConfigInstance()
            .setProperty(
                "hystrix.command.default"
                + ".circuitBreaker.errorThresholdPercentage",
                50)       // 错误率 50% 触发熔断
            ;
        ConfigurationManager.getConfigInstance()
            .setProperty(
                "hystrix.command.default"
                + ".circuitBreaker.sleepWindowInMilliseconds",
                5000)     // 熔断后 5 秒尝试恢复
            ;
        ConfigurationManager.getConfigInstance()
            .setProperty(
                "hystrix.command.default"
                + ".execution.isolation"
                + ".thread.timeoutInMilliseconds",
                3000)     // 请求超时 3 秒
            ;
    }
}

/**
 * 自适应并发限制示例
 * 基于 TCP Vegas 算法的思想
 */
public class AdaptiveConcurrencyLimiter {

    private volatile int currentLimit;
    private final int minLimit;
    private final int maxLimit;
    private final double smoothing;
    private volatile long minRtt = Long.MAX_VALUE;

    public AdaptiveConcurrencyLimiter(int minLimit, 
                                       int maxLimit,
                                       double smoothing) {
        this.minLimit = minLimit;
        this.maxLimit = maxLimit;
        this.smoothing = smoothing;
        this.currentLimit = minLimit;
    }

    /**
     * 根据请求的实际耗时动态调整并发限制
     * 核心思想：当延迟上升时减少并发，延迟下降时增加并发
     */
    public synchronized void onRequestComplete(
            long rttNanos, boolean didDrop) {
        if (didDrop) {
            currentLimit = Math.max(
                minLimit, (int) (currentLimit * 0.9));
            return;
        }

        // 更新最小 RTT（反映无排队时的理想延迟）
        minRtt = Math.min(minRtt, rttNanos);

        // 计算队列长度估计值
        int queueSize = (int) Math.ceil(
            currentLimit * (1.0 - (double) minRtt / rttNanos));

        int newLimit;
        if (queueSize < currentLimit / 2) {
            // 队列不饱和，可以增加并发
            newLimit = currentLimit + 1;
        } else if (queueSize > currentLimit) {
            // 队列积压，减少并发
            newLimit = (int) (currentLimit * 0.9);
        } else {
            newLimit = currentLimit;
        }

        // 平滑调整
        currentLimit = (int) (currentLimit * (1 - smoothing) 
                + newLimit * smoothing);
        currentLimit = Math.max(
            minLimit, Math.min(maxLimit, currentLimit));
    }

    public int getCurrentLimit() {
        return currentLimit;
    }
}

"""
OCA 内容调度逻辑（简化版）
根据地区的观看历史和预测数据决定内容分发策略
"""
from dataclasses import dataclass, field
from typing import List, Dict
from datetime import datetime


@dataclass
class VideoContent:
    video_id: str
    title: str
    size_bytes: int
    popularity_score: float
    region_scores: Dict[str, float] = field(
        default_factory=dict
    )


@dataclass
class OCANode:
    node_id: str
    location: str
    capacity_bytes: int
    used_bytes: int
    stored_videos: List[str] = field(
        default_factory=list
    )

    @property
    def available_bytes(self) -> int:
        return self.capacity_bytes - self.used_bytes


class ContentScheduler:
    """
    内容调度器：决定哪些视频应该被缓存到哪些 OCA 节点
    """

    def __init__(self, popularity_predictor):
        self.popularity_predictor = popularity_predictor

    def compute_fill_plan(
            self,
            videos: List[VideoContent],
            oca_nodes: List[OCANode]
    ) -> Dict[str, List[str]]:
        """
        计算内容填充计划
        返回格式：{oca_node_id: [video_id_1, video_id_2, ...]}
        """
        fill_plan: Dict[str, List[str]] = {
            node.node_id: [] for node in oca_nodes
        }

        for node in oca_nodes:
            # 预测该地区未来 24 小时的观看热度
            predicted_demand = (
                self.popularity_predictor.predict(
                    region=node.location,
                    time_window_hours=24
                )
            )

            # 按预测热度排序视频
            ranked_videos = sorted(
                videos,
                key=lambda v: predicted_demand.get(
                    v.video_id, 0),
                reverse=True
            )

            # 贪心填充：优先放置最热门的内容
            remaining_capacity = node.available_bytes
            for video in ranked_videos:
                if video.video_id in node.stored_videos:
                    continue
                if video.size_bytes <= remaining_capacity:
                    fill_plan[node.node_id].append(
                        video.video_id)
                    remaining_capacity -= video.size_bytes

                if remaining_capacity <= 0:
                    break

        return fill_plan

    def should_evict(
            self,
            node: OCANode,
            video_id: str,
            current_time: datetime
    ) -> bool:
        """
        判断是否应该从节点淘汰某个视频
        基于最近 7 天的实际观看数据
        """
        recent_views = (
            self.popularity_predictor.get_recent_views(
                video_id=video_id,
                region=node.location,
                days=7
            )
        )
        # 如果 7 天内几乎无人观看，标记为可淘汰
        return recent_views < 10

/**
 * Netflix 推荐系统在线层简化架构
 */
public class RecommendationEngine {

    private final CandidateGenerator candidateGenerator;
    private final FeatureStore featureStore;
    private final RankingModel rankingModel;
    private final FilterService filterService;

    public RecommendationEngine(
            CandidateGenerator candidateGenerator,
            FeatureStore featureStore,
            RankingModel rankingModel,
            FilterService filterService) {
        this.candidateGenerator = candidateGenerator;
        this.featureStore = featureStore;
        this.rankingModel = rankingModel;
        this.filterService = filterService;
    }

    /**
     * 为用户生成个性化推荐列表
     */
    public RecommendationResponse recommend(
            String userId, RecommendationContext context) {

        // 第一步：获取用户特征
        UserFeatures userFeatures = 
                featureStore.getUserFeatures(userId);

        // 第二步：候选集生成（多路召回）
        List<VideoCandidate> candidates = 
                candidateGenerator.generate(
                    userId, userFeatures, context);

        // 第三步：过滤（去除已看、地区限制等）
        List<VideoCandidate> filtered = 
                filterService.applyFilters(
                    userId, candidates, context);

        // 第四步：精排（使用深度学习模型打分）
        List<ScoredVideo> scored = rankingModel.rank(
            userFeatures, filtered);

        // 第五步：多样性重排（避免推荐结果过于单一）
        List<ScoredVideo> diversified = 
                applyDiversityBoost(scored);

        // 第六步：按行（Row）组织结果
        return assembleRows(diversified, context);
    }

    /**
     * 多样性提升：确保推荐结果覆盖多种类型
     */
    private List<ScoredVideo> applyDiversityBoost(
            List<ScoredVideo> scored) {
        Map<String, Integer> genreCounts = new HashMap<>();
        List<ScoredVideo> result = new ArrayList<>();

        for (ScoredVideo video : scored) {
            String genre = video.getPrimaryGenre();
            int count = genreCounts.getOrDefault(genre, 0);

            // 同一类型出现过多时降低得分权重
            double diversityPenalty = 
                    Math.pow(0.8, count);
            video.adjustScore(diversityPenalty);

            genreCounts.put(genre, count + 1);
            result.add(video);
        }

        result.sort(Comparator.comparingDouble(
            ScoredVideo::getAdjustedScore).reversed());
        return result;
    }
}

/**
 * Netflix A/B 测试分流逻辑（简化版）
 */
public class ABTestAllocator {

    private final ExperimentRegistry registry;

    public ABTestAllocator(ExperimentRegistry registry) {
        this.registry = registry;
    }

    /**
     * 确定用户在指定实验中属于哪个分组
     * 使用确定性哈希保证同一用户始终分到同一组
     */
    public String getAllocation(String userId, 
                                String experimentId) {
        Experiment experiment = registry.getExperiment(
            experimentId);
        if (experiment == null || !experiment.isActive()) {
            return "control";
        }

        // 确定性哈希：用户 ID + 实验 ID → 固定的分组
        long hash = deterministicHash(
            userId + ":" + experimentId);
        double bucket = (hash % 10000) / 10000.0;

        double cumulative = 0.0;
        for (ExperimentGroup group : 
                experiment.getGroups()) {
            cumulative += group.getTrafficPercentage();
            if (bucket < cumulative) {
                return group.getName();
            }
        }

        return "control";
    }

    private long deterministicHash(String input) {
        // MurmurHash3 确保均匀分布
        return Math.abs(MurmurHash3.hash64(
            input.getBytes()));
    }
}

/**
 * 事件驱动的区域故障自动检测和切换（改进后的方案）
 */
public class RegionalFailoverOrchestrator {

    private final HealthAggregator healthAggregator;
    private final TrafficShifter trafficShifter;
    private final AlertService alertService;
    private final double failoverThreshold;

    public RegionalFailoverOrchestrator(
            HealthAggregator healthAggregator,
            TrafficShifter trafficShifter,
            AlertService alertService,
            double failoverThreshold) {
        this.healthAggregator = healthAggregator;
        this.trafficShifter = trafficShifter;
        this.alertService = alertService;
        this.failoverThreshold = failoverThreshold;
    }

    /**
     * 持续监控并自动触发区域故障转移
     */
    public void monitorAndFailover(String region) {
        RegionHealth health = healthAggregator
                .getRegionHealth(region);

        double healthScore = health.getOverallScore();

        if (healthScore < failoverThreshold) {
            // 自动确认：连续三次检查都低于阈值
            boolean confirmed = confirmFailure(
                region, 3, Duration.ofSeconds(10));

            if (confirmed) {
                alertService.notifyOnCall(
                    "区域 " + region 
                    + " 健康度低于阈值，自动触发故障转移"
                );

                // 执行流量切换
                List<String> healthyRegions = 
                        getHealthyRegions(region);
                trafficShifter.shiftTraffic(
                    region, healthyRegions);

                // 持续监控源区域，等待恢复
                scheduleRecoveryCheck(region);
            }
        }
    }

    private boolean confirmFailure(
            String region, int requiredCount,
            Duration interval) {
        int failureCount = 0;
        for (int i = 0; i < requiredCount; i++) {
            RegionHealth health = healthAggregator
                    .getRegionHealth(region);
            if (health.getOverallScore() 
                    < failoverThreshold) {
                failureCount++;
            }
            try {
                Thread.sleep(interval.toMillis());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }
        return failureCount >= requiredCount;
    }

    private List<String> getHealthyRegions(
            String excludeRegion) {
        return healthAggregator.getAllRegions().stream()
                .filter(r -> !r.equals(excludeRegion))
                .filter(r -> healthAggregator
                        .getRegionHealth(r)
                        .getOverallScore() > 0.8)
                .collect(Collectors.toList());
    }

    private void scheduleRecoveryCheck(String region) {
        // 每分钟检查一次源区域是否恢复
        // 恢复后自动将流量切回
    }
}