springcloud gateway扩展支持多版本灰度

news2024/11/23 13:16:29

改造要求

需要在原有的调度策略中通过客户端header中的version进行1个服务多实例下进行二次分组，让指定的version在指定的版本实例下进行轮训调度。

需要改造的点

1.业务服务在发布到naocs中的元数据需要指定版本号

2.网关的调度策略中需要增加版本的区分

3.无版本的header。默认还继续使用轮训调度

4.使用jenkins进行发布服务的时候需要允许发布多个服务，

5.nacos中体现的形式是一个服务多个实例，只是元数据不同而已。

开始改造

在业务服务中的bootstrap.yml中指定version

gateway的改造

1.先定义好自己的LoadBanlancer

public class EnhancedRoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer {

    private final String serviceId;
    private final AtomicInteger position;

    private final ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

    public EnhancedRoundRobinLoadBalancer(String serviceId,
                                          ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider) {
        this.serviceId = serviceId;
        this.position = new AtomicInteger(new Random().nextInt(1000));
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
    }

    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        ServiceInstanceListSupplier supplier = this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
        return supplier.get(request).next().map((serviceInstances) -> {
            return this.processInstanceResponse(request, supplier, serviceInstances);
        });

    }

    /**
     * copy自 RoundRobinLoadBalancer
     *
     * @param request
     * @param supplier
     * @param serviceInstances {@link  RoundRobinLoadBalancer}
     * @return
     */
    private Response<ServiceInstance> processInstanceResponse(Request request,
                                                              ServiceInstanceListSupplier supplier,
                                                              List<ServiceInstance> serviceInstances) {
//        log.info("-----------------------------------------------------");
//        log.info("获取到的实例个数:{}", serviceInstances.size());
        //根据header头携带version进行同一个服务下实例过滤，过滤后再使用roundRobinInstance轮训
        String headerVersion = getRequestHeader(request, "version");
//        log.info("获取到的version:{}", headerVersion);
        if (!ObjectUtils.isEmpty(headerVersion)){
            ArrayList<ServiceInstance> versionServiceInstances = new ArrayList<>();
            for (ServiceInstance serviceInstance : serviceInstances) {
                Map<String, String> metadata = serviceInstance.getMetadata();
                String version = metadata.get("version");
                if (!ObjectUtils.isEmpty(version) && headerVersion.equals(version)){
                    versionServiceInstances.add(serviceInstance);
                }
            }
            //带版本的实例不为空的情况下，使用版本实例进行轮训
            if (!ObjectUtils.isEmpty(versionServiceInstances)){
//                log.info("使用版本实例轮训，携带版本实例个数：{}",versionServiceInstances.size());
                Response<ServiceInstance> serviceInstanceResponse = getRoundRobinInstance(versionServiceInstances);
                if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
                    ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
                }
                return serviceInstanceResponse;
            }
        }



        Response<ServiceInstance> serviceInstanceResponse = getRoundRobinInstance(serviceInstances);
        if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
            ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
        }
        return serviceInstanceResponse;
    }


    /**
     * 使用RoundRobin机制获取节点
     *
     * @param instances 实例
     * @return {@link Response }<{@link ServiceInstance }>
     * @author : baohui
     */
    private Response<ServiceInstance> getRoundRobinInstance(List<ServiceInstance> instances) {
        if (instances.isEmpty()) {
            if (log.isWarnEnabled()) {
                log.warn("不存在可用的服务: " + serviceId);
            }
            return new EmptyResponse();
        } else if (instances.size() == 1) {
            return new DefaultResponse(instances.get(0));
        } else {
            // 每一次计数器都自动+1，实现轮询的效果
            int pos = this.position.incrementAndGet() & Integer.MAX_VALUE;
            ServiceInstance instance = instances.get(pos % instances.size());
            return new DefaultResponse(instance);
        }
    }

    private String getRequestHeader(Request request, String headerField) {
        HttpHeaders headers = ((RequestDataContext) request.getContext()).getClientRequest().getHeaders();
        //log.info("headers:{}", headers);
        return headers.getFirst(headerField);


    }