微服務之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);
    }