我們都知道ribbon 是做負載均衡的一個組件逼纸，那么它是如何拿到 注冊中心的服務(wù)列表的呢拘哨？本次我們就來解決這個問題倘要。首先我們來看一個簡單的案例锈拨，使用ribbon 客戶端來實現(xiàn)一下負載均衡。
@Autowired// ribbon 客戶端
private LoadBalancerClient loadBalancerClient;
@GetMapping("/hello2")
public String hello2() {
RestTemplate restTemplate = new RestTemplate();
// 獲取服務(wù)實例
ServiceInstance serviceInstance = loadBalancerClient.choose("server-provider");
String url = String.format("http://%s:%s",serviceInstance.getHost(), serviceInstance.getPort() + "/hello");
String template = restTemplate.getForObject(url, String.class);
return template;
}
ServiceInstance serviceInstance = loadBalancerClient.choose("server-provider"); 這句話就是拿到 經(jīng)過負載均衡算法后 得到的一個 服務(wù)實例昭雌。進去看一下這個方法复唤。
*
* @author Ryan Baxter
*/
public interface ServiceInstanceChooser {
/**
* Chooses a ServiceInstance from the LoadBalancer for the specified service.
* @param serviceId The service ID to look up the LoadBalancer.
* @return A ServiceInstance that matches the serviceId.
*/
ServiceInstance choose(String serviceId);
}
- 去到 RibbonLoadBalancerClient
....
@Override
public ServiceInstance choose(String serviceId) {
//去到下面這個 方法
return choose(serviceId, null);
}
/**
* New: Select a server using a 'key'.
* @param serviceId of the service to choose an instance for
* @param hint to specify the service instance
* @return the selected {@link ServiceInstance}
*/
public ServiceInstance choose(String serviceId, Object hint) {
Server server = getServer(getLoadBalancer(serviceId), hint);
if (server == null) {
return null;
}
return new RibbonServer(serviceId, server, isSecure(server, serviceId),
serverIntrospector(serviceId).getMetadata(server));
}
Server server = getServer(getLoadBalancer(serviceId), hint);
- getLoadBalancer(serviceId): 看下這個方法。獲取一個 LoadBalancer
public class RibbonLoadBalancerClient implements LoadBalancerClient {
private SpringClientFactory clientFactory;
public RibbonLoadBalancerClient(SpringClientFactory clientFactory) {
this.clientFactory = clientFactory;
}
protected ILoadBalancer getLoadBalancer(String serviceId) {
return this.clientFactory.getLoadBalancer(serviceId);
}
}
- SpringClientFactory
/**
* Get the load balancer associated with the name.
* @param name name to search by
* @return {@link ILoadBalancer} instance
* @throws RuntimeException if any error occurs
*/
public ILoadBalancer getLoadBalancer(String name) {
return getInstance(name, ILoadBalancer.class);
}
@Override
public <C> C getInstance(String name, Class<C> type) {
C instance = super.getInstance(name, type);
if (instance != null) {
return instance;
}
IClientConfig config = getInstance(name, IClientConfig.class);
return instantiateWithConfig(getContext(name), type, config);
}
- C instance = super.getInstance(name, type);
C instance = super.getInstance(name, type); 這句往上走烛卧，就是從spring 容器中 獲取一個 instance -> ILoadBalancer 佛纫。這里就不往下追了。我們?nèi)タ匆幌?ILoadBalancer 這個接口总放。
- ILoadBalancer
public interface ILoadBalancer {
/**
* Initial list of servers.
* This API also serves to add additional ones at a later time
* The same logical server (host:port) could essentially be added multiple times
* (helpful in cases where you want to give more "weightage" perhaps ..)
*
* @param newServers new servers to add
*/
public void addServers(List<Server> newServers);
/**
* Choose a server from load balancer.
*
* @param key An object that the load balancer may use to determine which server to return. null if
* the load balancer does not use this parameter.
* @return server chosen
*/
public Server chooseServer(Object key);
/**
* To be called by the clients of the load balancer to notify that a Server is down
* else, the LB will think its still Alive until the next Ping cycle - potentially
* (assuming that the LB Impl does a ping)
*
* @param server Server to mark as down
*/
public void markServerDown(Server server);
/**
* @deprecated 2016-01-20 This method is deprecated in favor of the
* cleaner {@link #getReachableServers} (equivalent to availableOnly=true)
* and {@link #getAllServers} API (equivalent to availableOnly=false).
*
* Get the current list of servers.
*
* @param availableOnly if true, only live and available servers should be returned
*/
@Deprecated
public List<Server> getServerList(boolean availableOnly);
/**
* @return Only the servers that are up and reachable.
*/
public List<Server> getReachableServers();
/**
* @return All known servers, both reachable and unreachable.
*/
public List<Server> getAllServers();
}
這個接口有一個實現(xiàn)類 ：DynamicServerListLoadBalancer呈宇。
C instance = super.getInstance(name, type); 我們debug的話 會發(fā)現(xiàn) 這里的 instance -> DynamicServerListLoadBalancer。也就說這里返回的是 DynamicServerListLoadBalancer 對象局雄。
- DynamicServerListLoadBalancer
public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer {
private static final Logger LOGGER = LoggerFactory.getLogger(DynamicServerListLoadBalancer.class);
boolean isSecure = false;
boolean useTunnel = false;
// to keep track of modification of server lists
protected AtomicBoolean serverListUpdateInProgress = new AtomicBoolean(false);
volatile ServerList<T> serverListImpl;
volatile ServerListFilter<T> filter;
protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
@Override
public void doUpdate() {
updateListOfServers(); //更新服務(wù)列表甥啄。
}
};
@Deprecated
public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,
ServerList<T> serverList, ServerListFilter<T> filter) {
this(
clientConfig,
rule,
ping,
serverList,
filter,
new PollingServerListUpdater()
);
}
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);
}
public DynamicServerListLoadBalancer(IClientConfig clientConfig) {
initWithNiwsConfig(clientConfig);
}
- restOfInit(clientConfig); // initWithNiwsConfig(clientConfig);
initWithNiwsConfig(clientConfig) ; 這個方法里面會 調(diào)用 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(); // 開啟后臺線程 更新服務(wù)列表。
updateListOfServers(); // 更新服務(wù)列表炬搭。
if (primeConnection && this.getPrimeConnections() != null) {
this.getPrimeConnections()
.primeConnections(getReachableServers());
}
this.setEnablePrimingConnections(primeConnection);
LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
}
- enableAndInitLearnNewServersFeature();
/**
* Feature that lets us add new instances (from AMIs) to the list of
* existing servers that the LB will use Call this method if you want this
* feature enabled
*/
public void enableAndInitLearnNewServersFeature() {
LOGGER.info("Using serverListUpdater {}", serverListUpdater.getClass().getSimpleName());
serverListUpdater.start(updateAction); //開始執(zhí)行后臺更新蜈漓。
}
現(xiàn)在我們來看一下 serverListUpdater.start(updateAction); 前面有一段 下面的代碼：
protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
@Override
public void doUpdate() {
updateListOfServers(); //
}
};
// serverListUpdater.start(updateAction); // 這句話就是 回調(diào) 的 這里的 doUpdate()；
- serverListUpdater.start(updateAction); -> PollingServerListUpdater 走的這里的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(); // 回調(diào)融虽。
lastUpdated = System.currentTimeMillis();
} catch (Exception e) {
logger.warn("Failed one update cycle", e);
}
}
};
scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
wrapperRunnable, //傳入線程，定時輪詢灼芭，
initialDelayMs, //延遲啟動
refreshIntervalMs,// 服務(wù)列表刷新 時間間隔
TimeUnit.MILLISECONDS
);
} else {
logger.info("Already active, no-op");
}
}
到這里我們就明白了 有额，ribbon 也是 定時輪詢服務(wù)列表，使用后臺線程 完成的。
接下來我們 就來分析 一下 具體 獲取服務(wù)列表的方法 updateListOfServers();
- updateListOfServers();
@VisibleForTesting
public void updateListOfServers() {
List<T> servers = new ArrayList<T>();
if (serverListImpl != null) {
// 獲取服務(wù)列表
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);
}
這里呢巍佑，我們知道 ribbon 獲取了 服務(wù)列表后 保存在了本地緩存里面茴迁。 updateAllServerList(servers);
servers = serverListImpl.getUpdatedListOfServers();
serverListImpl -> DiscoveryEnabledNIWSServerList
@Override
public List<DiscoveryEnabledServer> getUpdatedListOfServers(){
return obtainServersViaDiscovery();
}
- 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>();
}
// 到這里 就明白了 ribbon 是通過 EurekaClient 來完成的服務(wù)列表的獲取，也就說使用的還是 eureka 自己的服務(wù)發(fā)現(xiàn)句狼。
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
// 獲取服務(wù)實例
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 = createServer(ii, isSecure, shouldUseIpAddr);
serverList.add(des);
}
}
if (serverList.size()>0 && prioritizeVipAddressBasedServers){
break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
}
}
}
// 返回了一個服務(wù)列表笋熬。
return serverList;
}
好了，到這里 我們就分析出 ribbon 是 如何 去 注冊中心獲取 服務(wù)列表的了腻菇。 ribbon 會在 后臺 開一個定時任務(wù)胳螟，使用 eureka client 自己的服務(wù)發(fā)現(xiàn) 去獲取 服務(wù)列表，然后 保存到本地緩存筹吐，當客戶段在 消費服務(wù)的時候 經(jīng)過負載均衡算法 選取其中一個服務(wù)實例糖耸，并通過 restTemplate 發(fā)送http 請求 ，請求到 服務(wù)提供者丘薛。完成帶有負載均衡的 遠程通信嘉竟。
這里我們會發(fā)現(xiàn)一個問題，也是在 使用spring cloud 開發(fā)中經(jīng)常 碰到的一個現(xiàn)象洋侨，就是 當服務(wù)在做集群的時候 舍扰，新增 或者 掛掉一個服務(wù)時候，不能及時反映到系統(tǒng)里面希坚。到這里我們就知道 原因了边苹，1. eureka client 端的 服務(wù)發(fā)現(xiàn) 是通過 后臺定時任務(wù)去更新的，保存到本地緩存的 裁僧，2. ribbon 也是通過 后臺定時任務(wù) 借助 于 eureka client 端的 服務(wù)發(fā)現(xiàn) 去查詢服務(wù)列表个束，然后保存到本地的。這里就產(chǎn)生了 延遲的問題聊疲。當然 eureka 本身是 居于 AP 的茬底，所以如果要使用 eureka 做為注冊中的話，一定要 明確 系統(tǒng)是否對 時效 性有要求获洲，如果有的話 就不適合阱表，可以 考慮 zk（CP） 、nacos 等贡珊。
服務(wù)列表我們拿到了最爬，接下來會 分析 ribbon 的負載均衡策略。
