spring cloud gateway 自定义负载均衡

小灰灰 2023-07-02 03:16 27阅读 0赞

--------------------

### spring cloud gateway 自定义负载均衡 ###

## ##

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**相关类及接口**

**LoadbalancerClientFilter：使用ribbon负载均衡，默认使用该类（已不推荐使用）**

/** @deprecated */
    @Deprecated
    public class LoadBalancerClientFilter implements GlobalFilter, Ordered {
        public static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10100;
        private static final Log log = LogFactory.getLog(LoadBalancerClientFilter.class);
        protected final LoadBalancerClient loadBalancer;
        private LoadBalancerProperties properties;
    
        public LoadBalancerClientFilter(LoadBalancerClient loadBalancer, LoadBalancerProperties properties) {
            this.loadBalancer = loadBalancer;
            this.properties = properties;
        }
    
        public int getOrder() {
            return 10100;
        }
    
        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 && ("lb".equals(url.getScheme()) || "lb".equals(schemePrefix))) {
                ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
                if (log.isTraceEnabled()) {
                    log.trace("LoadBalancerClientFilter url before: " + url);
                }
    
                ServiceInstance instance = this.choose(exchange);
                if (instance == null) {
                    throw NotFoundException.create(this.properties.isUse404(), "Unable to find instance for " + url.getHost());
                } else {
                    URI uri = exchange.getRequest().getURI();
                    String overrideScheme = instance.isSecure() ? "https" : "http";
                    if (schemePrefix != null) {
                        overrideScheme = url.getScheme();
                    }
    
                    URI requestUrl = this.loadBalancer.reconstructURI(new DelegatingServiceInstance(instance, overrideScheme), uri);
                    if (log.isTraceEnabled()) {
                        log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
                    }
    
                    exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl);
                    return chain.filter(exchange);
                }
            } else {
                return chain.filter(exchange);
            }
        }
    
        protected ServiceInstance choose(ServerWebExchange exchange) {
            return this.loadBalancer.choose(((URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR)).getHost());
        }
    }

**说明：默认使用该类，可通过下述方法使用ReactiveLoadbalancerClientFilter**

"You already have RibbonLoadBalancerClient on your classpath. It will be used by default. 
    As Spring Cloud Ribbon is in maintenance mode. We recommend switching to " + BlockingLoadBalancerClient.class.getSimpleName() + " instead. 
    In order to use it, set the value of `spring.cloud.loadbalancer.ribbon.enabled` to `false` 
    or remove spring-cloud-starter-netflix-ribbon from your project."

**ReactiveLoadBalancerClientFilter：负载均衡拦截器**

public class ReactiveLoadBalancerClientFilter 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 ReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) {
            this.clientFactory = clientFactory;
            this.properties = properties;
        }
    
        public int getOrder() {
            return 10150;
        }
    
        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 && ("lb".equals(url.getScheme()) || "lb".equals(schemePrefix))) {
                                //url不为null且协议为lb，或者url以lb开头
    
                ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url);
                if (log.isTraceEnabled()) {
                    log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url);
                }
    
                return this.choose(exchange).doOnNext((response) -> {
                                //获取ServiceInstance实例，进行一些处理
    
                    if (!response.hasServer()) {
                                //如果没有serviceInstance，直接抛出异常
    
                        throw NotFoundException.create(this.properties.isUse404(), "Unable to find instance for " + url.getHost());
                    } else {    //如果有serviceInstance，进行相关处理
                        URI uri = exchange.getRequest().getURI();
                        String overrideScheme = null;
                        if (schemePrefix != null) {
                            overrideScheme = url.getScheme();
                        }
    
                        DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance((ServiceInstance)response.getServer(), overrideScheme);
                        URI requestUrl = LoadBalancerUriTools.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); //如果获取不到serviceInstance，直接进行后续过滤
            }
        }
    
        private Mono<Response<ServiceInstance>> choose(ServerWebExchange exchange) {
            URI uri = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
            ReactorLoadBalancer<ServiceInstance> loadBalancer = (ReactorLoadBalancer)this.clientFactory.getInstance(uri.getHost(), ReactorLoadBalancer.class, new Class[]{ServiceInstance.class});
            if (loadBalancer == null) {
                throw new NotFoundException("No loadbalancer available for " + uri.getHost());
            } else {
                return loadBalancer.choose(this.createRequest());
            }
        }//选择服务实例
    
        private Request createRequest() {
            return ReactiveLoadBalancer.REQUEST;
        }
    }

**ReactorLoadBalancer：负载均衡接口**

public interface ReactorLoadBalancer<T> extends ReactiveLoadBalancer<T> {
        Mono<Response<T>> choose(Request request);
    
        default Mono<Response<T>> choose() {
            return this.choose(REQUEST);
        }
    }
    
    ***********************
    
    public interface ReactorServiceInstanceLoadBalancer extends ReactorLoadBalancer<ServiceInstance> {
    }

**RoundRobinLoadbalancer：负载均衡使用轮询**

public class RoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer {
        private static final Log log = LogFactory.getLog(RoundRobinLoadBalancer.class);
        private final AtomicInteger position;
    
        private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;
        private final String serviceId;
    
    ************
    构造方法
    
        public RoundRobinLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId) {
        public RoundRobinLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId, int seedPosition) {
    
    ************
    普通方法
    
        public Mono<Response<ServiceInstance>> choose(Request request) {
            if (this.serviceInstanceListSupplierProvider != null) {
                ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);
                return ((Flux)supplier.get()).next().map(this::getInstanceResponse);
            } else {
                ServiceInstanceSupplier supplier = (ServiceInstanceSupplier)this.serviceInstanceSupplier.getIfAvailable(NoopServiceInstanceSupplier::new);
                return ((Flux)supplier.get()).collectList().map(this::getInstanceResponse);
            }
        }
    
        private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
            if (instances.isEmpty()) {
                log.warn("No servers available for service: " + this.serviceId);
                return new EmptyResponse();
            } else {
                int pos = Math.abs(this.position.incrementAndGet());
                ServiceInstance instance = (ServiceInstance)instances.get(pos % instances.size());
                return new DefaultResponse(instance);
            }
        }//使用轮询获取实例
    }

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**示例：参数id为偶数时，输出hello new version**

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**网关**

**配置文件**

spring:
      application:
        name: hello-gateway
      cloud:
        consul:
          host: 172.18.0.20
          port: 8500
        loadbalancer:
          ribbon:
            enabled: false
        gateway:
          routes:
            - id: myRoute
              uri: lb://hello-service
              predicates:
                - Path=/hello

**自定义过滤器**

@Component
    public class CustomLoadBalancerClientFilter implements GlobalFilter, Ordered {
        private static final Log log = LogFactory.getLog(org.springframework.cloud.gateway.filter.ReactiveLoadBalancerClientFilter.class);
    
        @Resource
        private final LoadBalancerClientFactory clientFactory;
    
        @Resource
        private LoadBalancerProperties properties;
    
        public CustomLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) {
            this.clientFactory = clientFactory;
            this.properties = properties;
        }
    
        public int getOrder() {
            return 10149;
        }
    
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
            URI url = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
            String schemePrefix = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR);
            if (url != null && ("lb".equals(url.getScheme()) || "lb".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();
                        }
    
                        int id=Integer.parseInt(Objects.requireNonNull(exchange.getRequest().getQueryParams().getFirst("id")));
                        if (id%2==0){
                            while (!"new".equals(response.getServer().getMetadata().get("version"))){
                                try {
                                    response=this.choose(exchange).toFuture().get();
                                }catch (Exception e){
                                    System.out.println(e.getMessage());
                                }
                            }
                        }
    
                        DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance(response.getServer(), overrideScheme);
    
                        System.out.println(exchange.getRequest().getQueryParams().getFirst("id")+"对应server的version为："+serviceInstance.getMetadata().get("version"));
                        URI requestUrl = LoadBalancerUriTools.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);
            }
        }
    
        private Mono<Response<ServiceInstance>> choose(ServerWebExchange exchange) {
            URI uri = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
            assert uri != null;
            ReactorLoadBalancer<ServiceInstance> loadBalancer = this.clientFactory.getInstance(uri.getHost(), ReactorLoadBalancer.class, new Class[]{ServiceInstance.class});
            if (loadBalancer == null) {
                throw new NotFoundException("No loadbalancer available for " + uri.getHost());
            } else {
                return loadBalancer.choose(this.createRequest());
            }
        }
    
        private Request createRequest() {
            return ReactiveLoadBalancer.REQUEST;
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**同名应用hello-service1**

**配置文件**

spring:
      application:
        name: hello-service
      cloud:
        consul:
          host: 172.18.0.20
          port: 8500
          discovery:
            instance-id: ${spring.application.name}-${random.int}
            tags: version=old

**controller 层**

@RestController
    public class HelloController {
    
        @RequestMapping("/hello")
        public String hello(){
            return "hello old version";
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**同名应用hello-service2**

**配置文件**

**controller 层**

@RestController
    public class HelloController {
    
        @RequestMapping("/hello")
        public String hello(){
            return "hello new version";
        }
    }

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**测试输出**

**consul注册的应用**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

**参数测试**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1][]

**当id为偶数时，输出为hello new version**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200127215114701.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200128095709240.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ==,size_16,color_FFFFFF,t_70