Java流量切换案例

本文目录导读：

1. 流量切换方案设计
2. 完整实现案例

我来为您提供一个完整的Java流量切换（Traffic Switching）案例，这是微服务架构中常见的技术需求。

流量切换方案设计

核心方案对比

// 1. 基于权重随机切换
// 2. 基于版本号切换  
// 3. 基于请求特征路由
// 4. 基于动态配置中心切换

完整实现案例

1 基础模型定义

import lombok.Data;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
// 流量路由配置
@Data
public class TrafficRouteConfig {
    private String serviceName;           // 服务名
    private Map<String, Integer> weights; // 版本 -> 权重
    private String defaultVersion;        // 默认版本
    private boolean enableGray;           // 是否启用灰度
    private String grayRule;              // 灰度规则
    public static TrafficRouteConfig defaultConfig() {
        TrafficRouteConfig config = new TrafficRouteConfig();
        config.setServiceName("order-service");
        config.setDefaultVersion("v1");
        config.setEnableGray(false);
        config.setWeights(new ConcurrentHashMap<>());
        return config;
    }
}

2 流量切换核心处理器

import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@Component
public class TrafficSwitchHandler {
    private final Map<String, TrafficRouteConfig> routeConfigs = new ConcurrentHashMap<>();
    private final SecureRandom random = new SecureRandom();
    // 基于权重的路由选择
    public String selectVersionByWeight(String serviceName) {
        TrafficRouteConfig config = routeConfigs.getOrDefault(
            serviceName, TrafficRouteConfig.defaultConfig()
        );
        if (!config.isEnableGray()) {
            return config.getDefaultVersion();
        }
        Map<String, Integer> weights = config.getWeights();
        if (weights.isEmpty()) {
            return config.getDefaultVersion();
        }
        // 计算总权重
        int totalWeight = weights.values().stream()
            .mapToInt(Integer::intValue)
            .sum();
        if (totalWeight <= 0) {
            return config.getDefaultVersion();
        }
        // 随机选择
        int randomWeight = random.nextInt(totalWeight);
        int current = 0;
        for (Map.Entry<String, Integer> entry : weights.entrySet()) {
            current += entry.getValue();
            if (randomWeight < current) {
                return entry.getKey();
            }
        }
        return config.getDefaultVersion();
    }
    // 基于请求特征的灰度路由
    public String selectVersionByRule(String serviceName, String userId, String requestParams) {
        TrafficRouteConfig config = routeConfigs.get(serviceName);
        if (config == null || !config.isEnableGray()) {
            return "v1"; // 默认版本
        }
        // 示例灰度规则：特定用户ID走新版本
        if (userId != null && userId.startsWith("test_")) {
            return "v2"; // 测试用户走v2
        }
        // 示例灰度规则：随机百分比
        if (shouldRouteToNewVersion(requestParams)) {
            return "v2";
        }
        return config.getDefaultVersion();
    }
    private boolean shouldRouteToNewVersion(String params) {
        // 基于请求参数的规则判断
        if (params != null && params.contains("new_feature")) {
            return true;
        }
        // 基于时间或其他策略
        return random.nextDouble() < 0.1; // 10%流量
    }
    // 更新流量配置
    public void updateRouteConfig(TrafficRouteConfig config) {
        routeConfigs.put(config.getServiceName(), config);
    }
}

3 动态配置中心集成

import com.alibaba.nacos.api.config.annotation.NacosConfigListener;
import com.alibaba.nacos.api.config.annotation.NacosConfigurationProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
@NacosConfigurationProperties(dataId = "traffic-route-config", autoRefreshed = true)
public class DynamicTrafficConfig {
    private volatile String routeConfig;
    @Autowired
    private TrafficSwitchHandler trafficSwitchHandler;
    @Autowired
    private ObjectMapper objectMapper;
    @NacosConfigListener(dataId = "traffic-route-config")
    public void onConfigChange(String configJson) {
        try {
            // 解析配置并更新
            TrafficRouteConfig[] configs = objectMapper.readValue(
                configJson, TrafficRouteConfig[].class
            );
            for (TrafficRouteConfig config : configs) {
                trafficSwitchHandler.updateRouteConfig(config);
            }
        } catch (Exception e) {
            System.err.println("解析流量配置失败: " + e.getMessage());
        }
    }
    public String getRouteConfig() {
        return routeConfig;
    }
}

4 业务调用示例

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
@RestController
@RequestMapping("/api/order")
public class OrderController {
    @Autowired
    private TrafficSwitchHandler trafficSwitchHandler;
    @Autowired
    private ServiceInvoker serviceInvoker;
    @PostMapping("/create")
    public Order createOrder(@RequestBody OrderRequest request, 
                              @RequestHeader(value = "X-User-ID", required = false) String userId) {
        // 获取目标版本
        String targetVersion = trafficSwitchHandler.selectVersionByRule(
            "order-service", userId, request.toString()
        );
        // 根据版本调用不同服务
        if ("v2".equals(targetVersion)) {
            return serviceInvoker.invokeV2(request, targetVersion);
        } else {
            return serviceInvoker.invokeV1(request, targetVersion);
        }
    }
    @GetMapping("/stats")
    public Map<String, Long> getTrafficStats() {
        return serviceInvoker.getTrafficStats();
    }
}
// 服务调用器
@Component
public class ServiceInvoker {
    private final Map<String, AtomicInteger> versionCounters = new ConcurrentHashMap<>();
    public Order invokeV1(OrderRequest request, String version) {
        incrementCounter(version);
        // 调用V1版本服务
        return new Order(version, "order-001");
    }
    public Order invokeV2(OrderRequest request, String version) {
        incrementCounter(version);
        // 调用V2版本服务
        return new Order(version, "order-001-v2");
    }
    private void incrementCounter(String version) {
        AtomicInteger counter = versionCounters.computeIfAbsent(
            version, k -> new AtomicInteger(0)
        );
        counter.incrementAndGet();
    }
    public Map<String, Long> getTrafficStats() {
        Map<String, Long> stats = new HashMap<>();
        versionCounters.forEach((version, counter) -> {
            stats.put(version, counter.longValue());
        });
        return stats;
    }
}

5 配置管理接口

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping("/admin/traffic")
public class TrafficAdminController {
    @Autowired
    private TrafficSwitchHandler trafficSwitchHandler;
    // 查询当前配置
    @GetMapping("/config/{serviceName}")
    public TrafficRouteConfig getConfig(@PathVariable String serviceName) {
        return trafficSwitchHandler.getConfig(serviceName);
    }
    // 更新权重配置
    @PostMapping("/weight")
    public void updateWeight(@RequestBody WeightUpdateRequest request) {
        TrafficRouteConfig config = trafficSwitchHandler.getConfig(request.getServiceName());
        if (config != null) {
            // 更新权重比例
            config.getWeights().put(request.getVersion(), request.getWeight());
            // 重新计算总权重为100
            normalizeWeights(config);
            trafficSwitchHandler.updateRouteConfig(config);
        }
    }
    // 设置灰度比例
    @PostMapping("/gray")
    public void setGrayRatio(@RequestParam String serviceName, 
                              @RequestParam int ratio) {
        TrafficRouteConfig config = trafficSwitchHandler.getConfig(serviceName);
        if (config != null) {
            config.setEnableGray(ratio > 0);
            // 设置新版本权重
            config.getWeights().put("v2", ratio);
            config.getWeights().put("v1", 100 - ratio);
            trafficSwitchHandler.updateRouteConfig(config);
        }
    }
    private void normalizeWeights(TrafficRouteConfig config) {
        Map<String, Integer> weights = config.getWeights();
        int total = weights.values().stream()
            .mapToInt(Integer::intValue)
            .sum();
        if (total != 100 && total > 0) {
            // 归一化到100
            weights.replaceAll((key, value) -> 
                (int) Math.round(value * 100.0 / total)
            );
        }
    }
}
// 请求体定义
@Data
public class WeightUpdateRequest {
    private String serviceName;
    private String version;
    private int weight;
}

6 配置示例

# application.yml
spring:
  application:
    name: order-service
traffic:
  route:
    config:
      - service: order-service
        default: v1
        enable-gray: true
        weight:
          v1: 90
          v2: 10
        rules:
          - type: header
            key: X-User-Type
            values: ["test", "beta"]
            target: v2
          - type: percentage
            ratio: 0.1
            target: v2

7 使用场景对比

8 最佳实践建议

// 1. 提供降级机制
public class TrafficFallbackHandler {
    // 熔断降级
    @SentinelResource(value = "traffic-switch", fallback = "fallback")
    public String handleTraffic() {
        // 流量切换逻辑
    }
    public String fallback() {
        // 降级到默认版本
        return "v1";
    }
}
// 2. 监控埋点
public class TrafficMonitor {
    // 实时监控各版本流量
    public void trackTraffic(String service, String version) {
        // 发送到监控系统
        TagMetric metric = new TagMetric("traffic")
            .addField("count", 1)
            .addTag("service", service)
            .addTag("version", version);
        // 发送到Prometheus/Grafana
        metricsCollector.record(metric);
    }
}

这个流量切换案例涵盖了：

• 基于权重的路由策略
• 基于特征的灰度路由
• 动态配置更新
• 监控和统计
• 管理接口

实际使用时可根据具体业务场景调整策略,结合服务网格（如Istio）或API网关实现更高级的流量管理。