基於springcloud gateway + nacos實現灰度發佈（reactive版）

時間 2020-06-29

標籤基於 springcloud gateway nacos 實現灰度發佈 reactive 欄目 CGI 简体版

原文原文鏈接

什麼是灰度發佈？

灰度發佈（又名金絲雀發佈）是指在黑與白之間，可以平滑過渡的一種發佈方式。在其上能夠進行A/B testing，即讓一部分用戶繼續用產品特性A，一部分用戶開始用產品特性B，若是用戶對B沒有什麼反對意見，那麼逐步擴大範圍，把全部用戶都遷移到B上面來。灰度發佈能夠保證總體系統的穩定，在初始灰度的時候就能夠發現、調整問題，以保證其影響度。javascript

本文以springcloud gateway + nacos來演示如何實現灰度發佈，若是對springcloud gateway和nacos還不熟悉的朋友，能夠先閱讀以下文章，而後再閱讀本文。html

springcloud gateway官方介紹java

nacos官方介紹react

實現的總體思路：git

編寫帶權重的灰度路由
編寫自定義filter
nacos服務配置須要灰度發佈的服務的元數據信息以及權重
灰度路由從nacos服務拉取元數據信息以及權重，而後根據權重算法，返回符合要求的服務實例給自定義的filter
網關配置文件配置須要灰度路由的服務（由於本文代碼沒有網關實現動態路由，否則灰度路由能夠配置在配置中心，從配置中心拉取）
filter經過責任鏈模式，把服務實例透傳給其餘filter好比NettyRoutingFilter

下邊進入實戰github

正文

一、所使用的開發版本web

<jdk.version>1.8</jdk.version>
    <!-- spring cloud -->
    <spring-cloud.version>Hoxton.SR3</spring-cloud.version>
    <spring-boot.version>2.2.5.RELEASE</spring-boot.version>
    <spring-cloud-alibaba.version>2.2.1.RELEASE</spring-cloud-alibaba.version>

二、pom.xml引入算法

<dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-gateway</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.cloud</groupId>
                    <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-loadbalancer</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>

    </dependencies>

ps：nacos的jar注意排除ribbon依賴，否則loadbalancer沒法生效spring

三、編寫權重路由apache

public class GrayLoadBalancer implements ReactorServiceInstanceLoadBalancer {
    private static final Log log = LogFactory.getLog(GrayLoadBalancer.class);
    private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
    private  String serviceId;




    public GrayLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId) {
        this.serviceId = serviceId;
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
    }


    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        HttpHeaders headers = (HttpHeaders) request.getContext();
        if (this.serviceInstanceListSupplierProvider != null) {
            ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
            return ((Flux)supplier.get()).next().map(list->getInstanceResponse((List<ServiceInstance>)list,headers));
        }

        return null;


    }



    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances,HttpHeaders headers) {
        if (instances.isEmpty()) {
            return getServiceInstanceEmptyResponse();
        } else {
            return getServiceInstanceResponseWithWeight(instances);
        }
    }

    /**
     * 根據版本進行分發
     * @param instances
     * @param headers
     * @return
     */
    private Response<ServiceInstance> getServiceInstanceResponseByVersion(List<ServiceInstance> instances, HttpHeaders headers) {
        String versionNo = headers.getFirst("version");
        System.out.println(versionNo);
        Map<String,String> versionMap = new HashMap<>();
        versionMap.put("version",versionNo);
        final Set<Map.Entry<String,String>> attributes =
                Collections.unmodifiableSet(versionMap.entrySet());
        ServiceInstance serviceInstance = null;
        for (ServiceInstance instance : instances) {
            Map<String,String> metadata = instance.getMetadata();
            if(metadata.entrySet().containsAll(attributes)){
                serviceInstance = instance;
                break;
            }
        }

        if(ObjectUtils.isEmpty(serviceInstance)){
            return getServiceInstanceEmptyResponse();
        }
        return new DefaultResponse(serviceInstance);
    }

    /**
     *
     * 根據在nacos中配置的權重值，進行分發
     * @param instances
     *
     * @return
     */
    private Response<ServiceInstance> getServiceInstanceResponseWithWeight(List<ServiceInstance> instances) {
        Map<ServiceInstance,Integer> weightMap = new HashMap<>();
        for (ServiceInstance instance : instances) {
            Map<String,String> metadata = instance.getMetadata();
            System.out.println(metadata.get("version")+"-->weight:"+metadata.get("weight"));
            if(metadata.containsKey("weight")){
                weightMap.put(instance,Integer.valueOf(metadata.get("weight")));
            }
        }
        WeightMeta<ServiceInstance> weightMeta = WeightRandomUtils.buildWeightMeta(weightMap);
        if(ObjectUtils.isEmpty(weightMeta)){
            return getServiceInstanceEmptyResponse();
        }
        ServiceInstance serviceInstance = weightMeta.random();
        if(ObjectUtils.isEmpty(serviceInstance)){
            return getServiceInstanceEmptyResponse();
        }
        System.out.println(serviceInstance.getMetadata().get("version"));
        return new DefaultResponse(serviceInstance);
    }

    private Response<ServiceInstance> getServiceInstanceEmptyResponse() {
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    }

四、自定義filter

public class GrayReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered {

    private static final Log log = LogFactory.getLog(ReactiveLoadBalancerClientFilter.class);
    private static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10150;
    private final LoadBalancerClientFactory clientFactory;
    private LoadBalancerProperties properties;

    public GrayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) {
        this.clientFactory = clientFactory;
        this.properties = properties;
    }

    @Override
    public int getOrder() {
        return LOAD_BALANCER_CLIENT_FILTER_ORDER;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        URI url = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
        String schemePrefix = (String)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR);
        if (url != null && ("grayLb".equals(url.getScheme()) || "grayLb".equals(schemePrefix))) {
            ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
            if (log.isTraceEnabled()) {
                log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url);
            }

            return this.choose(exchange).doOnNext((response) -> {
                if (!response.hasServer()) {
                    throw NotFoundException.create(this.properties.isUse404(), "Unable to find instance for " + url.getHost());
                } else {
                    URI uri = exchange.getRequest().getURI();
                    String overrideScheme = null;
                    if (schemePrefix != null) {
                        overrideScheme = url.getScheme();
                    }

                    DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance((ServiceInstance)response.getServer(), overrideScheme);
                    URI requestUrl = this.reconstructURI(serviceInstance, uri);
                    if (log.isTraceEnabled()) {
                        log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
                    }

                    exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl);
                }
            }).then(chain.filter(exchange));
        } else {
            return chain.filter(exchange);
        }
    }

    protected URI reconstructURI(ServiceInstance serviceInstance, URI original) {
        return LoadBalancerUriTools.reconstructURI(serviceInstance, original);
    }

    private Mono<Response<ServiceInstance>> choose(ServerWebExchange exchange) {
        URI uri = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
        GrayLoadBalancer loadBalancer = new GrayLoadBalancer(clientFactory.getLazyProvider(uri.getHost(), ServiceInstanceListSupplier.class), uri.getHost());
        if (loadBalancer == null) {
            throw new NotFoundException("No loadbalancer available for " + uri.getHost());
        } else {
            return loadBalancer.choose(this.createRequest(exchange));
        }
    }

    private Request createRequest(ServerWebExchange exchange) {
        HttpHeaders headers = exchange.getRequest().getHeaders();
        Request<HttpHeaders> request = new DefaultRequest<>(headers);
        return request;
    }
}

五、配置自定義filter給spring管理

@Configuration
public class GrayGatewayReactiveLoadBalancerClientAutoConfiguration {
    public GrayGatewayReactiveLoadBalancerClientAutoConfiguration() {
    }



    @Bean
    @ConditionalOnMissingBean({GrayReactiveLoadBalancerClientFilter.class})
    public GrayReactiveLoadBalancerClientFilter grayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) {
        return new GrayReactiveLoadBalancerClientFilter(clientFactory, properties);
    }




}

六、編寫網關application.yml配置

server:
  port: 9082
# 配置輸出日誌
logging:
  level:
    org.springframework.cloud.gateway: TRACE
    org.springframework.http.server.reactive: DEBUG
    org.springframework.web.reactive: DEBUG
    reactor.ipc.netty: DEBUG

#開啓端點
management:
  endpoints:
    web:
      exposure:
        include: '*'
spring:
  application:
    name: gateway-reactor-gray
  cloud:
     nacos:
       discovery:
        server-addr: localhost:8848
     gateway:
       discovery:
         locator:
           enabled: true
           lower-case-service-id: true
       routes:
         - id: hello-consumer
           uri: grayLb://hello-consumer
           predicates:
              - Path=/hello/**

uri中的grayLb配置，表明該服務須要進行灰度發佈

七、在註冊中心nacos配置灰度發佈的服務版本以及權重值

weight表明權重，version表明版本

總結

上述就是實現灰度發佈的過程，實現灰度發佈的方法有不少種，文章中只是提供一種思路。雖然springcloud官方推薦使用loadbalancer來代替ribbon。由於ribbon是阻塞的，但從官方的loadbalancer的負載均衡算法來看，目前loadbalancer默認只支持輪詢算法，要其餘算法得本身擴展實現，而ribbon默認支持7種算法，用默認的算法基本上就能夠知足咱們的需求了。其次ribbon支持懶加載處理，超時以及重試與斷路器hystrix集成等配置，loadbalancer目前就支持重試。因此若是正式環境要本身實現灰度發佈，能夠考慮對ribbon進行擴展。本文的實現只是做爲一種擴展補充，畢竟springcloud推薦loadbalancer，索性就寫個demo實現下。

最後灰度發佈的實現，業內也有開源的實現--Discovery，感興趣的朋友能夠經過以下連接進行查看