Spring Cloud Ribbon 源碼解析

時間 2020-03-03

標籤 spring cloud ribbon 源碼解析欄目 Spring 简体版

原文原文鏈接

專欄目錄

Spring Cloud OpenFeign 源碼解析
Spring Cloud Ribbon 源碼解析
Spring Cloud Alibaba Sentinel 源碼解析
Spring Cloud Gatway 源碼解析
Spring Cloud Alibaba Nacos 源碼解析

代碼準備

依賴關係

+------------+              +------------+
|            |              |            |
|            |              |            |
|            |              |            |
|            |              |            |
|  consumer  +------------> |   provider |
|            | RestTemplate |            |
|            |              |            |
|            |              |            |
|            |              |            |
+------------+              +------------+

pom 依賴

加入nacos 服務發現便可，內部引用了 spring-cloud-ribbon 相關依賴java

<dependency>
  <groupId>com.alibaba.cloud</groupId>
  <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

調用客戶端

咱們這裏以最簡單的 RestTemplate 調用開始使用Ribbongit

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

// Controller 使用restTemplate 調用服務提供方接口
ResponseEntity<String> forEntity = restTemplate.getForEntity("http://provider/req", String.class);

源碼解析

建立調用攔截器

1. 獲取所有 `@LoadBalanced`標記的`RestTemplate`

public class LoadBalancerAutoConfiguration {
    @LoadBalanced
    @Autowired(required = false)
    private List<RestTemplate> restTemplates = Collections.emptyList();
}

2. 增長 `LoadBalancerInterceptor` 處理邏輯

沒有引入 spring-retry使用的是

@Bean
public LoadBalancerInterceptor ribbonInterceptor() {
    return new LoadBalancerInterceptor();
}

引入 spring-retry 使用的是

@Bean
@ConditionalOnMissingBean
public RetryLoadBalancerInterceptor ribbonInterceptor() {
    return new RetryLoadBalancerInterceptor();
}

LoadBalancerInterceptor 業務邏輯

public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
    @Override
    public ClientHttpResponse intercept() {
        final URI originalUri = request.getURI();
        // http://demo-provider/req 截取 demo-provider 服務名稱
        String serviceName = originalUri.getHost();

        // 默認注入的 RibbonAutoConfiguration.RibbonLoadBalancerClient
        return this.loadBalancer.execute(serviceName,
                // 建立請求對象
                this.requestFactory.createRequest(request, body, execution));
    }
}

執行攔截器

3. RibbonLoadBalancerClient執行

//RibbonAutoConfiguration默認注入的RibbonLoadBalancerClient
@Bean
@ConditionalOnMissingBean(LoadBalancerClient.class)
public LoadBalancerClient loadBalancerClient() {
  return new RibbonLoadBalancerClient(springClientFactory());
}

4.execute執行

public class RibbonLoadBalancerClient implements LoadBalancerClient {
    public <T> T execute(){
        //獲取具體的ILoadBalancer實現
        ILoadBalancer loadBalancer = getLoadBalancer(serviceId);

        // 調用ILoadBalancer 實現獲取Server
        Server server = getServer(loadBalancer, hint);
        RibbonServer ribbonServer = new RibbonServer(serviceId, server,
                isSecure(server, serviceId),
                serverIntrospector(serviceId).getMetadata(server));

        //獲取狀態記錄器，保存這次選取的server
        RibbonLoadBalancerContext context = this.clientFactory
                .getLoadBalancerContext(serviceId);
        RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server);
        T returnVal = request.apply(serviceInstance);
        statsRecorder.recordStats(returnVal);
        return returnVal;
    }
}

獲取ILoadBalancer

5 SpringClientFactory

// bean 工廠生成LoadBalancer 的實現
protected ILoadBalancer getLoadBalancer(String serviceId) {
    return this.springClientFactory.getLoadBalancer(serviceId);
}

// 具體生成邏輯看 RibbonClientConfiguration，這個Bean 只有工廠調用的時候纔會建立
@Bean
@ConditionalOnMissingBean
public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
    return new ZoneAwareLoadBalancer<>();
}

6.建立LoadBalancer 的依賴要素

名稱	默認實現	做用
IClientConfig	DefaultClientConfigImpl	ribbon 客戶端配置參數，例如：超時設置、壓縮設置等
ServerList	NacosServerList	目標服務的實例實例表，具體服務發現客戶端實現
ServerListFilter	ZonePreferenceServerListFilter	針對ServerList 實例列表的過濾邏輯處理
IRule	ZoneAvoidanceRule	負載均衡選擇Server 的規則
IPing	DummyPing	檢驗服務是否可用的方法實現
ServerListUpdater	PollingServerListUpdater	針對ServerList 更新的操做實現

以上默認實現參考 RibbonClientConfiguration. ZoneAwareLoadBalancergithub

獲取服務實例

//Server server = getServer(loadBalancer, hint);  4. excute 方法
protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
    return loadBalancer.chooseServer(hint != null ? hint : "default");
}

7. ZoneAwareLoadBalancer

public class ZoneAwareLoadBalancer{
    public ZoneAwareLoadBalancer() {
        // 調用父類初始化方法。 這裏會開啓實例維護的定時任務等 （具體解析參考 擴展部分）
        super(clientConfig, rule, ping, serverList, filter, serverListUpdater);
    }
    @Override
    public Server chooseServer(Object key) {
        // 如果使用的 Nacos 服務發現，則沒有 Zone 的概念，直接調用父類的實現
        if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
            return super.chooseServer(key);
        }
        // 如下爲有 Zone 的概念 例如 Eureka  （具體）
        ...
        return server;
    }
}

父類調用IRule實現選擇Server

public Server chooseServer(Object key) {
    return rule.choose(key);
}

8.PredicateBasedRule 選擇規則

public abstract class PredicateBasedRule {
    @Override
    public Server choose(Object key) {
        ILoadBalancer lb = getLoadBalancer();
        // 獲取斷言配置
        Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
        if (server.isPresent()) {
            return server.get();
        } else {
            return null;
        }
    }
}

9. ZoneAvoidancePredicate服務列表斷言

public class ZoneAvoidancePredicate {
    @Override
    public boolean apply(@Nullable PredicateKey input) {
        if (!ENABLED.get()) {
            return true;
        }
        // 仍是獲取區域配置，如是使用的 Nacos 直接返回true
        String serverZone = input.getServer().getZone();
        if (serverZone == null) {
            // there is no zone information from the server, we do not want to filter
            // out this server
            return true;
        }
        // 區域高可用判斷
        ...
    }
}

擴展: ServerList 維護

初始化ServerList

在上文 6.建立LoadBalancer 的依賴要素，中 ServerList 目標服務的實例實例表，具體服務發現客戶端實現。咱們來看下 Nacos 的實現spring

public class NacosServerList extends AbstractServerList<NacosServer> {
    @Override
    public List<NacosServer> getInitialListOfServers() {
        return getServers();
    }

    @Override
    public List<NacosServer> getUpdatedListOfServers() {
        return getServers();
    }

    private List<NacosServer> getServers() {
        String group = discoveryProperties.getGroup();
        //調用nacos-sdk 查詢實例列表
        List<Instance> instances = discoveryProperties.namingServiceInstance()
                .selectInstances(serviceId, group, true);
        // 類型轉換
        return instancesToServerList(instances);

    }
}

更新ServerListUpdater

ServerList 初始化後更新操做經過 PollingServerListUpdater

public class PollingServerListUpdater implements ServerListUpdater {
    @Override
    public synchronized void start(final UpdateAction updateAction) {
        // 更新任務 交給updateAction 具體實現
        final Runnable wrapperRunnable = () -> {
            updateAction.doUpdate();
            lastUpdated = System.currentTimeMillis();
        };

        // 開啓後臺線程定時執行  updateAction
        scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                wrapperRunnable,
                initialDelayMs,
                refreshIntervalMs,
                TimeUnit.MILLISECONDS
        );
    }
}

updateAction 實現

public void doUpdate() {
    DynamicServerListLoadBalancer.this.updateListOfServers();
}

public class PollingServerListUpdater implements ServerListUpdater {
    public void updateListOfServers() {
        List<T> servers = new ArrayList();
        
        // 調用NacosServiceList 獲取所有服務列表
        servers = this.serverListImpl.getUpdatedListOfServers();
        
        // 若是配置實例過濾器在執行過濾
        if (this.filter != null) {
            servers = this.filter.getFilteredListOfServers((List)servers);
        }
        
        // 更新LoadBalancer 服務列表
        this.updateAllServerList((List)servers);
    }
}

擴展: Server 狀態維護

LoadBalancer 初始構造時會觸發 setupPingTask()

public BaseLoadBalancer() {
  this.name = DEFAULT_NAME;
  this.ping = null;
  setRule(DEFAULT_RULE);
  // 開啓ping 檢查任務
  setupPingTask();
  lbStats = new LoadBalancerStats(DEFAULT_NAME);
}

setupPingTask

void setupPingTask() {
  // 是否能夠ping, 默認的DummyPing 直接 跳過不執行
  if (canSkipPing()) {
    return;
  }
  // 執行PingTask
  lbTimer.schedule(new BaseLoadBalancer.PingTask(), 0, pingIntervalSeconds * 1000);
  // 開啓任務
  new BaseLoadBalancer.Pinger(pingStrategy).runPinger();
}

SerialPingStrategy 串行執行邏輯

// 串行調度執行 Iping 邏輯
private static class SerialPingStrategy implements IPingStrategy {
  @Override
  public boolean[] pingServers(IPing ping, Server[] servers) {
    int numCandidates = servers.length;
    boolean[] results = new boolean[numCandidates];

    for (int i = 0; i < numCandidates; i++) {
      results[i] = false; /* Default answer is DEAD. */
      if (ping != null) {
        results[i] = ping.isAlive(servers[i]);
      }

    }
    return results;
  }
}

調用url 判斷可用性

public class PingUrl implements IPing {
    public boolean isAlive(Server server) {
        urlStr = urlStr + server.getId();
        urlStr = urlStr + this.getPingAppendString();
        boolean isAlive = false;
        HttpClient httpClient = new DefaultHttpClient();
        HttpUriRequest getRequest = new HttpGet(urlStr);
        String content = null;

        HttpResponse response = httpClient.execute(getRequest);
        content = EntityUtils.toString(response.getEntity());
        isAlive = response.getStatusLine().getStatusCode() == 200;
        return isAlive;
    }
}

擴展: RibbonClient 懶加載處理

由上文可知，默認狀況下 Ribbon 在第一次請求才會去建立 LoadBalancer ，這種懶加載機制會致使服務啓動後，第一次調用服務延遲問題，甚至在整合斷路器（hystrix）等出現超時熔斷。segmentfault

爲了解決這個問題，咱們會配置 Ribbon 的飢餓加載架構

ribbon:
  eager-load:
    clients:
      - provider

RibbonApplicationContextInitializer 服務啓動後自動調用工廠提早建立須要的ribbon clients

public class RibbonApplicationContextInitializer
        implements ApplicationListener<ApplicationReadyEvent> {
    private final List<String> clientNames;

    protected void initialize() {
        if (clientNames != null) {
            for (String clientName : clientNames) {
                this.springClientFactory.getContext(clientName);
            }
        }
    }
    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        initialize();
    }

}