微服务之RIBBON(实现自定义加权负载均衡策略)

客户端负载均衡RIBBON

• 概念
• 基本原理
• Ribbon 实现负载均衡
• 切换负载均衡策略
• 自定义负载均衡策略

概念

-----------------------------------完整流程概述-----------------------------------

整个流程就是 spring通过扫描所有添加了 @LoadBalanced 的 RestTemplate 给他们创建

LoadBalancerInterceptor

这个 拦截器中引入了

RibbonLoadBalancerClient

.

在 RestTemplate 执行方法的时候会调用 拦截器,其中又调用了 client 的 execute 方法. 方法中会通过serverId 创建一个

ILoadBalancer

其中包含了服务的配置信息,和

IRule

. 最后通过 IRule 的 choose 方法选择出一个合适的服务.

-----------------------------------完整流程概述-----------------------------------

git地址 : eureka集群+ribbon负载均衡模式.

ribbon 是一套

客户端

负载均衡,在

注册中心(EUREKA,NACOS)

上获取服务注册列表后缓存到jvm本地,在本地实现RPC远程调用

maven坐标,

spring-cloud-starter-netflix-eureka-client

坐标自己依赖了ribbon

<dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
      <version>2.2.3.RELEASE</version>
      <scope>compile</scope>
    </dependency>
           

基本原理

由springboot 原理可知,springboot引入了

LoadBalancerAutoConfiguration

类

// 类路径上有RestTemplate
@ConditionalOnClass(RestTemplate.class)
// 容器中有LoadBalancerClient实例的时候
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {

	// 维护了所有被  @LoadBalanced 标记了的 resttemplate
	@LoadBalanced
	@Autowired(required = false)
	private List<RestTemplate> restTemplates = Collections.emptyList();

	@Bean
	@ConditionalOnMissingBean
	public LoadBalancerRequestFactory loadBalancerRequestFactory(
			LoadBalancerClient loadBalancerClient) {
		return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers);
	}

	@Configuration(proxyBeanMethods = false)
	@ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
	static class LoadBalancerInterceptorConfig {

// 		根据loadbalancerclient 创建一个拦截器
		@Bean
		public LoadBalancerInterceptor ribbonInterceptor(
				LoadBalancerClient loadBalancerClient,
				LoadBalancerRequestFactory requestFactory) {
			return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
		}

}
           

由上面的源码可知,spring创建了一个

LoadBalancerInterceptor

其作用是当 RestTemplate 在发送请求的时候会拦截到方法,执行intercept方法.

而

LoadBalancerInterceptor

中有一个

LoadBalancerClient

在 intercept 中其实就是执行 execute 方法

public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {

	private LoadBalancerClient loadBalancer;

	private LoadBalancerRequestFactory requestFactory;

	@Override
	public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
			final ClientHttpRequestExecution execution) throws IOException {
		final URI originalUri = request.getURI();
		String serviceName = originalUri.getHost();
		Assert.state(serviceName != null,
				"Request URI does not contain a valid hostname: " + originalUri);
		return this.loadBalancer.execute(serviceName,
				this.requestFactory.createRequest(request, body, execution));
	}

}
           

这个

LoadBalancerClient

其实就是 RibbonLoadBalancerClient 由以下源码可知

@Configuration
@Conditional(RibbonAutoConfiguration.RibbonClassesConditions.class)
@RibbonClients
// 在spring的loadbalancer配置加载之前加载,于是创建的LoadBalancerClient就是ribbon提供的
@AutoConfigureAfter(
		name = "org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration")
@AutoConfigureBefore({ LoadBalancerAutoConfiguration.class,
		AsyncLoadBalancerAutoConfiguration.class })
@EnableConfigurationProperties({ RibbonEagerLoadProperties.class,
		ServerIntrospectorProperties.class })
public class RibbonAutoConfiguration {

	@Bean
	@ConditionalOnMissingBean
	public SpringClientFactory springClientFactory() {
		SpringClientFactory factory = new SpringClientFactory();
		factory.setConfigurations(this.configurations);
		return factory;
	}

	@Bean
	@ConditionalOnMissingBean(LoadBalancerClient.class)
	public LoadBalancerClient loadBalancerClient() {
		return new RibbonLoadBalancerClient(springClientFactory());
	}
}
           

由此可知,每次RestTemplate 调用方法的时候都会执行

RibbonLoadBalancerClient

里的 execute 方法

public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
			throws IOException {
			// 1. 获取 ILoadBalancer
		ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
		// 2. 通过 ILoadBalancer 选择了一个server
		Server server = getServer(loadBalancer, hint);
		if (server == null) {
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		RibbonServer ribbonServer = new RibbonServer(serviceId, server,
				isSecure(server, serviceId),
				serverIntrospector(serviceId).getMetadata(server));

		return execute(serviceId, ribbonServer, request);
	}
           

ILoadBalancer 在

RibbonClientConfiguration

中定义了创建过程. 其中传入了

IClientConfig

定义Ribbon管理配置得接口 和

IRule

定义Ribbon中负载均衡策略的接口. 不同的 serverId 会有不同的 ILoadBalancer.

@Bean
	@ConditionalOnMissingBean
	public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
			ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
			IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
		if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
			return this.propertiesFactory.get(ILoadBalancer.class, config, name);
		}
		return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,
				serverListFilter, serverListUpdater);
	}
           

execute 方法的第一步,获得了 ILoadBalancer 后,其中包含了服务名称和对应的负载均衡策略.第二步

getServer

方法中, 就是调用了 loadBalancer 的chooseServer 方法

protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
		if (loadBalancer == null) {
			return null;
		}
		// Use 'default' on a null hint, or just pass it on?
		return loadBalancer.chooseServer(hint != null ? hint : "default");
	}
           

public Server chooseServer(Object key) {
        if (counter == null) {
            counter = createCounter();
        }
        counter.increment();
        if (rule == null) {
            return null;
        } else {
            try {
            //  找到rule执行 choose 方法获得节点
                return rule.choose(key);
            } catch (Exception e) {
                logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);
                return null;
            }
        }
    }
           

可以看到, ILoadBalancer 的 choose 方法,最终调用的是 IRule 的choose方法.

@Override
    public Server choose(Object key) {
        // 如果所有服务都用完了则重新拷贝一份

        ILoadBalancer loadBalancer = getLoadBalancer();
        List<Server> reachableServers = loadBalancer.getReachableServers();
        return reachableServers.get(0);
    }
           

接下来就是选择一台服务的事情了,也是通过 ILoadBalancer 来获得服务列表

负载均衡就是在这个方法中实现的

Ribbon 实现负载均衡

切换负载均衡策略

网上说的都是自定义一个配置文件还不能被springboot扫描到,其实没这么麻烦.

@EnableEurekaClient
@SpringBootApplication
// 只需要在这里指定服务名称和对应的负载均衡策略就可以了
@RibbonClients({
        @RibbonClient(name = "PROVIDER",configuration = MyRule.class),
        @RibbonClient(name = "PROVIDER-02",configuration = RandomRule.class)

})
public class ConsumerApp {

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

    @Bean
    @LoadBalanced
    public RestTemplate restTemplate(){
        return new RestTemplate();
    }
}
           

自定义负载均衡策略

实现

AbstractLoadBalancerRule

抽象方法, 切记不可在自定义方法上加注解到容器中!!!否则负载均衡会出问题!!

自定义加权负载均衡策略.实现线程无阻塞加权负载均衡,只能设置大概率,如果完全相同,则需要加锁,个人觉得性能不好

private static List<String> arr = new ArrayList<>();

    static {
        // 如果想要实现不重启更新配置，可以开启一个定时器刷新更新配置文件
        try {
            Properties properties = PropertiesLoaderUtils.loadAllProperties("ribbonRule.properties");
            Enumeration<?> enumeration = properties.propertyNames();
            while (enumeration.hasMoreElements()){
                String key = (String) enumeration.nextElement();
                String value = properties.getProperty(key);
                // 获得每个服务的占比
                Integer size = Integer.valueOf(value);
                for (Integer i = 0; i < size; i++) {
                    arr.add(key);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
    @Override
    public void initWithNiwsConfig(IClientConfig iClientConfig) {
        System.out.println("");
    }




    private static final Random random = new Random();
    @Override
    public Server choose(Object key) {
        // 如果所有服务都用完了则重新拷贝一份

        ILoadBalancer loadBalancer = getLoadBalancer();
        List<Server> reachableServers = loadBalancer.getReachableServers();
        int size = arr.size();
        int index = random.nextInt(size);
        String port = arr.get(index);
        Optional<Server> any = reachableServers.stream().filter(server -> port.equals(server.getHostPort())).findAny();
        return any.orElse(null);
    }