【一塊兒學源碼-微服務】Ribbon 源碼三：Ribbon與Eureka整合原理分析

時間  2020-02-01
標籤 一塊兒學源碼-微服務 ribbon 源碼 eureka 整合 原理 分析 简体版
原文   原文鏈接

前言

前情回顧

上一篇講了Ribbon的初始化過程，從LoadBalancerAutoConfiguration 到RibbonAutoConfiguration 再到RibbonClientConfiguration，咱們找到了ILoadBalancer默認初始化的對象等。java

本講目錄

這一講咱們會進一步往下探究Ribbon和Eureka是如何結合的。微信

經過上一講ILoadBalancer 咱們已經能夠拿到一個服務全部的服務節點信息了，這裏面是怎麼把服務的名稱轉化爲對應的具體host請求信息的呢？app

經過這一講 咱們來一探究竟less

目錄以下：async

  1. EurekaClientAutoConfiguration.getLoadBalancer()回顧
  2. 再次梳理Ribbon初始化過程
  3. ServerList實現類初始化過程
  4. getUpdatedListOfServers()獲取註冊表列表分析
  5. ribbon如何更新本身保存的註冊表信息？

說明

原創不易，如若轉載 請標明來源！ide

博客地址：一枝花算不算浪漫
微信公衆號：壹枝花算不算浪漫函數

源碼閱讀

EurekaClientAutoConfiguration.getLoadBalancer()回顧

上一講咱們已經深刻的講解過getLoadBalancer() 方法的實現，每一個serviceName都對應一個本身的SpringContext上下文信息，而後經過ILoadBalancer.class從上下文信息中獲取默認的LoadBalancer：ZoneAwareLoadBalancer， 咱們看下這個類的構造函數：學習

public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule,
                             IPing ping, ServerList<T> serverList, ServerListFilter<T> filter,
                             ServerListUpdater serverListUpdater) {
    super(clientConfig, rule, ping, serverList, filter, serverListUpdater);
}

繼續跟父類DynamicServerListLoadBalancer的初始化方法：ui

public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer {
    volatile ServerList<T> serverListImpl;

    volatile ServerListFilter<T> filter;

    public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,
                                         ServerList<T> serverList, ServerListFilter<T> filter,
                                         ServerListUpdater serverListUpdater) {
        super(clientConfig, rule, ping);
        this.serverListImpl = serverList;
        this.filter = filter;
        this.serverListUpdater = serverListUpdater;
        if (filter instanceof AbstractServerListFilter) {
            ((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());
        }
        restOfInit(clientConfig);
    }

    void restOfInit(IClientConfig clientConfig) {
        boolean primeConnection = this.isEnablePrimingConnections();
        // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
        this.setEnablePrimingConnections(false);
        enableAndInitLearnNewServersFeature();

        updateListOfServers();
        if (primeConnection && this.getPrimeConnections() != null) {
            this.getPrimeConnections()
                    .primeConnections(getReachableServers());
        }
        this.setEnablePrimingConnections(primeConnection);
        LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
    }

    @VisibleForTesting
    public void updateListOfServers() {
        List<T> servers = new ArrayList<T>();
        if (serverListImpl != null) {
            servers = serverListImpl.getUpdatedListOfServers();
            LOGGER.debug("List of Servers for {} obtained from Discovery client: {}",
                    getIdentifier(), servers);

            if (filter != null) {
                servers = filter.getFilteredListOfServers(servers);
                LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}",
                        getIdentifier(), servers);
            }
        }
        updateAllServerList(servers);
    }
}

構造方法中有個restOfInit()方法，進去後又會有updateListOfServers() 方法，看方法名就知道這個確定是和server註冊表相關的，繼續日後看，servers = serverListImpl.getUpdatedListOfServers();，這裏直接調用getUpdatedListOfServers()就獲取到了全部的註冊表信息。this

0.jpeg

能夠看到ServerList有四個實現類，這個究竟是該調用哪一個實現類的getUpdatedListOfServers()方法呢？接着往下看。

再次梳理Ribbon初始化過程

第二講咱們已經見過Ribbon的初始化過程，並畫了圖整理，這裏針對於以前的圖再更新一下：

這裏主要是增長了RibbonEurekaAutoConfiguration和EurekaRibbonClientConfiguration兩個配置類的初始化。

ServerList實現類初始化過程

上面已經梳理過 Ribbon初始化的過程，其中在EurekaRibbonClientConfiguration 會初始化RibbonServerList，代碼以下：

@Configuration
    public class EurekaRibbonClientConfiguration {
    @Bean
    @ConditionalOnMissingBean
    public ServerList<?> ribbonServerList(IClientConfig config, Provider<EurekaClient> eurekaClientProvider) {
        if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {
            return this.propertiesFactory.get(ServerList.class, config, serviceId);
        }
        DiscoveryEnabledNIWSServerList discoveryServerList = new DiscoveryEnabledNIWSServerList(
                config, eurekaClientProvider);
        DomainExtractingServerList serverList = new DomainExtractingServerList(
                discoveryServerList, config, this.approximateZoneFromHostname);
        return serverList;
    }
}

這裏實際的ServerList實際就是DiscoveryEnabledNIWSServerList，咱們看下這個類：

public class DiscoveryEnabledNIWSServerList extends AbstractServerList<DiscoveryEnabledServer>{

}

public abstract class AbstractServerList<T extends Server> implements ServerList<T>, IClientConfigAware {

}

因此能夠看出來ServerList 實際就是在這裏進行初始化的，上面那個serverListImpl.getUpdatedListOfServers();即爲調用DiscoveryEnabledNIWSServerList.getUpdatedListOfServers() 方法了，繼續往下看。

getUpdatedListOfServers()獲取註冊表分析

直接看DiscoveryEnabledNIWSServerList.getUpdatedListOfServers()源代碼：

@Override
public List<DiscoveryEnabledServer> getUpdatedListOfServers(){
    return obtainServersViaDiscovery();
}

private List<DiscoveryEnabledServer> obtainServersViaDiscovery() {
    List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>();

    if (eurekaClientProvider == null || eurekaClientProvider.get() == null) {
        logger.warn("EurekaClient has not been initialized yet, returning an empty list");
        return new ArrayList<DiscoveryEnabledServer>();
    }

    EurekaClient eurekaClient = eurekaClientProvider.get();
    if (vipAddresses!=null){
        for (String vipAddress : vipAddresses.split(",")) {
            // if targetRegion is null, it will be interpreted as the same region of client
            List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
            for (InstanceInfo ii : listOfInstanceInfo) {
                if (ii.getStatus().equals(InstanceStatus.UP)) {

                    if(shouldUseOverridePort){
                        if(logger.isDebugEnabled()){
                            logger.debug("Overriding port on client name: " + clientName + " to " + overridePort);
                        }

                        // copy is necessary since the InstanceInfo builder just uses the original reference,
                        // and we don't want to corrupt the global eureka copy of the object which may be
                        // used by other clients in our system
                        InstanceInfo copy = new InstanceInfo(ii);

                        if(isSecure){
                            ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build();
                        }else{
                            ii = new InstanceInfo.Builder(copy).setPort(overridePort).build();
                        }
                    }

                    DiscoveryEnabledServer des = new DiscoveryEnabledServer(ii, isSecure, shouldUseIpAddr);
                    des.setZone(DiscoveryClient.getZone(ii));
                    serverList.add(des);
                }
            }
            if (serverList.size()>0 && prioritizeVipAddressBasedServers){
                break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
            }
        }
    }
    return serverList;
}

看到這裏代碼就已經很明顯了，咱們來解讀下這段代碼：

  1. 經過eurekaClientProvider獲取對應EurekaClient
  2. 經過vipAdress(實際就是serviceName)獲取對應註冊表集合信息
  3. 將註冊信息組裝成DiscoveryEnabledServer列表

再回到DynamicServerListLoadBalancer.updateListOfServers() 中，這裏獲取到對應的DiscoveryEnabledServer list後調用updateAllServerList()方法，一路跟蹤這裏最終會調用BaseLoadBalancer.setServersList()

public class BaseLoadBalancer extends AbstractLoadBalancer implements
        PrimeConnections.PrimeConnectionListener, IClientConfigAware {

    @Monitor(name = PREFIX + "AllServerList", type = DataSourceType.INFORMATIONAL)
    protected volatile List<Server> allServerList = Collections
            .synchronizedList(new ArrayList<Server>());
            
    public void setServersList(List lsrv) {
        Lock writeLock = allServerLock.writeLock();
        logger.debug("LoadBalancer [{}]: clearing server list (SET op)", name);
        
        ArrayList<Server> newServers = new ArrayList<Server>();
        writeLock.lock();
        try {
            ArrayList<Server> allServers = new ArrayList<Server>();
            for (Object server : lsrv) {
                if (server == null) {
                    continue;
                }

                if (server instanceof String) {
                    server = new Server((String) server);
                }

                if (server instanceof Server) {
                    logger.debug("LoadBalancer [{}]:  addServer [{}]", name, ((Server) server).getId());
                    allServers.add((Server) server);
                } else {
                    throw new IllegalArgumentException(
                            "Type String or Server expected, instead found:"
                                    + server.getClass());
                }

            }
            boolean listChanged = false;
            if (!allServerList.equals(allServers)) {
                listChanged = true;
                if (changeListeners != null && changeListeners.size() > 0) {
                   List<Server> oldList = ImmutableList.copyOf(allServerList);
                   List<Server> newList = ImmutableList.copyOf(allServers);                   
                   for (ServerListChangeListener l: changeListeners) {
                       try {
                           l.serverListChanged(oldList, newList);
                       } catch (Exception e) {
                           logger.error("LoadBalancer [{}]: Error invoking server list change listener", name, e);
                       }
                   }
                }
            }
            if (isEnablePrimingConnections()) {
                for (Server server : allServers) {
                    if (!allServerList.contains(server)) {
                        server.setReadyToServe(false);
                        newServers.add((Server) server);
                    }
                }
                if (primeConnections != null) {
                    primeConnections.primeConnectionsAsync(newServers, this);
                }
            }
            // This will reset readyToServe flag to true on all servers
            // regardless whether
            // previous priming connections are success or not
            allServerList = allServers;
            if (canSkipPing()) {
                for (Server s : allServerList) {
                    s.setAlive(true);
                }
                upServerList = allServerList;
            } else if (listChanged) {
                forceQuickPing();
            }
        } finally {
            writeLock.unlock();
        }
    }
}

這個過程最後用一張圖總結爲：

ribbon如何更新本身保存的註冊表信息？

上面已經講了 Ribbon是如何經過serviceName拉取到註冊表的，咱們知道EurekaClient默認是30s拉取一次註冊表信息的，由於Ribbon要關聯註冊表信息，那麼Ribbon該如何更新本身存儲的註冊表信息呢？

繼續回到DynamicSeverListLoadBalancer.restOfInit()方法中：

public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer {

    protected volatile ServerListUpdater serverListUpdater;

    void restOfInit(IClientConfig clientConfig) {
        boolean primeConnection = this.isEnablePrimingConnections();
        // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
        this.setEnablePrimingConnections(false);
        enableAndInitLearnNewServersFeature();

        updateListOfServers();
        if (primeConnection && this.getPrimeConnections() != null) {
            this.getPrimeConnections()
                    .primeConnections(getReachableServers());
        }
        this.setEnablePrimingConnections(primeConnection);
        LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
    }

    public void enableAndInitLearnNewServersFeature() {
        LOGGER.info("Using serverListUpdater {}", serverListUpdater.getClass().getSimpleName());
        serverListUpdater.start(updateAction);
    }
}

重點查看enableAndInitLearnNewServersFeature()方法，從名字咱們就能夠看出來這意思爲激活和初始化學習新服務的功能，這裏實際上就啓動serverListUpdater中的一個線程。

在最上面Ribbon初始化的過程當中咱們知道，在RibbonClientConfiguration中默認初始化的ServerListUpdater爲 PollingServreListUpdater，咱們繼續跟這個類的start方法：

@Override
public synchronized void start(final UpdateAction updateAction) {
    if (isActive.compareAndSet(false, true)) {
        final Runnable wrapperRunnable = new Runnable() {
            @Override
            public void run() {
                if (!isActive.get()) {
                    if (scheduledFuture != null) {
                        scheduledFuture.cancel(true);
                    }
                    return;
                }
                try {
                    updateAction.doUpdate();
                    lastUpdated = System.currentTimeMillis();
                } catch (Exception e) {
                    logger.warn("Failed one update cycle", e);
                }
            }
        };

        scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                wrapperRunnable,
                initialDelayMs,
                refreshIntervalMs,
                TimeUnit.MILLISECONDS
        );
    } else {
        logger.info("Already active, no-op");
    }
}

這裏只要是執行updateAction.doUpdate();,而後後面啓動了一個調度任務，默認30s執行一次。

繼續日後跟doUpdate()方法：

public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer {
    protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
        @Override
        public void doUpdate() {
            updateListOfServers();
        }
    };
}

這裏又調用了以前經過serviceName獲取對應註冊服務列表的方法了。

總結到一張圖以下：

註冊表服務

總結

本文主要是從新梳理了Ribbon的初始化過程，主要是幾個Configure初始化的過程，而後是Ribbon與Eureka的整合，這裏也涉及到了註冊表的更新邏輯。

看到這裏真是被Spring的各類AutoConfigure繞暈了，哈哈，可是最後分析完 仍是以爲挺清晰的，對於複雜的業務畫張流程圖還挺容易理解的。

申明

本文章首發自本人博客：https://www.cnblogs.com/wang-meng 和公衆號：壹枝花算不算浪漫，如若轉載請標明來源!

感興趣的小夥伴可關注我的公衆號：壹枝花算不算浪漫

22.jpg

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程