这篇文章给大家介绍Ribbon中怎么使用 LoadBalancer 实现负载均衡,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。
Ribbon 的负载均衡器是通过 LoadBalancerClient 来实现的,在应用启动的时候,LoadBalancerClient 默认会从 EurekaClient 获取服务列表,并将服务注册列表缓存在本地,当调用 LoadBalancerClient 的 choose() 方法的时候, 根据负载均衡策略 IRule 来选择一个可用的服务,从而实现负载均衡。
当然,LoadBalancerClient 也可以不从 EurekaClient 中获取服务列表,这是需要自己维护一个服务注册列表信息,具体操作如下:
ribbon: eureka: enabled: false stores: ribbon: listOfServers: baidu.com, google.com
主要流程:
1. 当应用启动的时候,ILoadBalancer 从 EurekaClient 获取服务列表
2. 然后每 10 秒 向 EurekaClient 发送一次心跳检测,如果注册列表发生了变化,则更新获取重新获取
3. LoadBalancerClient 调用 choose() 方法来选择服务的时候,会调用 ILoadBalancer 的 chooseServer() 来获取一个可以的服务,
4. 在 ILoadBalancer 进行获取服务的时候,会根据负载均衡策略 IRule 来进行选择
5. 返回可用的服务
下面就来看看每个类的实现原理
RibbonLoadBalancerClient 它是 Ribbon 负载均衡实现的一个重要的类,最终的负载均衡的请求处理都由它来执行,先来看下它的类图:
它实现了 LoadBalancerClient 接口,而 LoadBalancerClient 接口实现了 ServiceInstanceChooser 接口:
该接口用来从负载均衡器中获取一个可用的服务,只有一个方法:
public interface ServiceInstanceChooser { /** * @param serviceId:服务ID * @return 可用的服务实例 */ ServiceInstance choose(String serviceId); }
表示负载均衡的客户端,是一个接口,继承了 ServiceInstanceChooser 接口 ,共有三个方法:
public interface LoadBalancerClient extends ServiceInstanceChooser { /** * 执行请求 * @param serviceId :用于查找 LoadBalancer的服务ID * @param request:允许实现执行前后操作 */ <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException; /** * 执行请求 * @param serviceId :用于查找 LoadBalancer的服务ID * @param serviceInstance :执行请求的服务 * @param request:允许实现执行前后操作 */ <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException; /** * 创建具有真实主机和端口的正确URI,有些系统使用带有逻辑服务名的URL作为主机,调用该方法将会使用 host:port 来替换逻辑服务名 * @param instance :用于重建URI的服务实例 * @param original :具有逻辑服务名的URL * @return A reconstructed URI. */ URI reconstructURI(ServiceInstance instance, URI original); }
主要看下从 ServiceInstanceChooser,LoadBalancerClient 中实现的接口方法
public class RibbonLoadBalancerClient implements LoadBalancerClient { // 工厂:主要用来创建客户端,创建负载均衡器,进行客户端配置等 // 对于每一个客户端名称都会创建一个Spring ApplicationContext,可以从中获取需要的bean private SpringClientFactory clientFactory; protected ILoadBalancer getLoadBalancer(String serviceId) { return this.clientFactory.getLoadBalancer(serviceId); } .................. } public class SpringClientFactory extends NamedContextFactory<RibbonClientSpecification> { // 获取客户端 public <C extends IClient<?, ?>> C getClient(String name, Class<C> clientClass) { return getInstance(name, clientClass); } // 获取负载均衡器 public ILoadBalancer getLoadBalancer(String name) { return getInstance(name, ILoadBalancer.class); } //获取客户端配置 public IClientConfig getClientConfig(String name) { return getInstance(name, IClientConfig.class); } // 获取 RibbonLoadBalancerContext public RibbonLoadBalancerContext getLoadBalancerContext(String serviceId) { return getInstance(serviceId, RibbonLoadBalancerContext.class); } // 获取对应的bean public <T> T getInstance(String name, Class<T> type) { AnnotationConfigApplicationContext context = getContext(name); ..... return context.getBean(type); ..... } }
该方法主要用来获取一个可用的服务实例
public ServiceInstance choose(String serviceId, Object hint) { Server server = getServer(getLoadBalancer(serviceId), hint); if (server == null) { return null; } // RibbonServer 实现了 ServiceInstance return new RibbonServer(serviceId, server, isSecure(server, serviceId), serverIntrospector(serviceId).getMetadata(server)); } // 根据服务ID获取负载均衡器,会调用 SpringClientFactory 的方法进行获取 protected ILoadBalancer getLoadBalancer(String serviceId) { return this.clientFactory.getLoadBalancer(serviceId); } // 根据负载均衡器来获取可用的服务 protected Server getServer(ILoadBalancer loadBalancer, Object hint) { if (loadBalancer == null) { return null; } return loadBalancer.chooseServer(hint != null ? hint : "default"); }
最后会调用 ILoadBalancer.chooseServer 来获取可用服务,后面再来说 ILoadBalancer 。
该方法执行请求
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException { Server server = null; if (serviceInstance instanceof RibbonServer) { server = ((RibbonServer) serviceInstance).getServer(); } RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId); // 状态记录器,记录着服务的状态 RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server); ........... T returnVal = request.apply(serviceInstance); statsRecorder.recordStats(returnVal); return returnVal; ........... } // apply 方法调用如下,最终返回 ClientHttpResponse public ListenableFuture<ClientHttpResponse> intercept(final HttpRequest request, final byte[] body, final AsyncClientHttpRequestExecution execution) throws IOException { final URI originalUri = request.getURI(); String serviceName = originalUri.getHost(); return this.loadBalancer.execute(serviceName, new LoadBalancerRequest<ListenableFuture<ClientHttpResponse>>() { @Override public ListenableFuture<ClientHttpResponse> apply( final ServiceInstance instance) throws Exception { HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, AsyncLoadBalancerInterceptor.this.loadBalancer); return execution.executeAsync(serviceRequest, body); } }); }
以上就是负载均衡器流程图左边部分的原理,接下来看下右边的部分
通过上面的分析,负载均衡器获取一个可用的服务,最终会调用 ILoadBalancer 的 chooseServer 方法,下面就来看下 ILoadBalancer 的实现原理
首先来看下 ILoadBalancer 的整体类图:
在上面的类图中,主要的逻辑实在 BaseLoadBalancer 中实现,而 DynamicServerListLoadBalancer 主要提供动态获取服务列表的能力。
首先来看下 ILoadBalancer,它表示一个负载均衡器接口,
public interface ILoadBalancer { // 添加服务 public void addServers(List<Server> newServers); //获取服务 public Server chooseServer(Object key); //标记某个服务下线 public void markServerDown(Server server); //获取状态为UP的所有可用服务列表 public List<Server> getReachableServers(); //获取所有服务列表,包括可用的和不可用的 public List<Server> getAllServers(); }
实现 ILoadBalancer 接口,提供一些默认实现
public abstract class AbstractLoadBalancer implements ILoadBalancer { public enum ServerGroup{ALL, STATUS_UP, STATUS_NOT_UP} public Server chooseServer() { return chooseServer(null); } public abstract List<Server> getServerList(ServerGroup serverGroup); // 获取状态 public abstract LoadBalancerStats getLoadBalancerStats(); }
客户端配置
public interface IClientConfigAware { public abstract void initWithNiwsConfig(IClientConfig clientConfig); }
负载均衡器的主要实现逻辑,在该类中,会根据负载均衡策略 IRule 来获取可用的服务,会通过 IPing 来检测服务的可用性,此外,该类中从 EurkaClient 获取到服务列表后,会在该类中保存下来,会维护所有的服务列表和可用的服务列表。
首先来看下它的一些属性,然后再来看每个对应的方法
public class BaseLoadBalancer extends AbstractLoadBalancer implements PrimeConnections.PrimeConnectionListener, IClientConfigAware { // 默认的负载均衡策略:轮询选择服务实例 private final static IRule DEFAULT_RULE = new RoundRobinRule(); protected IRule rule = DEFAULT_RULE; // 默认 ping 的策略,会调用 IPing 来检测服务是否可用 private final static SerialPingStrategy DEFAULT_PING_STRATEGY = new SerialPingStrategy(); protected IPingStrategy pingStrategy = DEFAULT_PING_STRATEGY; protected IPing ping = null; // 所有服务列表 protected volatile List<Server> allServerList = Collections.synchronizedList(new ArrayList<Server>()); // 状态为 up 的服务列表 protected volatile List<Server> upServerList = Collections.synchronizedList(new ArrayList<Server>()); // 锁 protected ReadWriteLock allServerLock = new ReentrantReadWriteLock(); protected ReadWriteLock upServerLock = new ReentrantReadWriteLock(); // 定时任务,去 ping 服务是否可用 protected Timer lbTimer = null; // ping 的时间间隔,10秒 protected int pingIntervalSeconds = 10; // ping 的最大次数 protected int maxTotalPingTimeSeconds = 5; // 负载均衡器的状态 protected LoadBalancerStats lbStats; // 客户端配置 private IClientConfig config; // 服务列表变化监听器 private List<ServerListChangeListener> changeListeners = new CopyOnWriteArrayList<ServerListChangeListener>(); // 服务状态变化监听器 private List<ServerStatusChangeListener> serverStatusListeners = new CopyOnWriteArrayList<ServerStatusChangeListener>(); // 构造方法,使用默认的配置来创建负载均衡器,还有其他重载的构造方法,可用根据需要来创建负载均衡器 public BaseLoadBalancer() { this.name = DEFAULT_NAME; this.ping = null; setRule(DEFAULT_RULE); setupPingTask(); lbStats = new LoadBalancerStats(DEFAULT_NAME); } ..................... }
在上面的属性中,Ribbon 提供了一些默认的配置:
IClientConfig 表示客户端的配置,实现类为 DefaultClientConfigImpl,在该类中配置了默认的值,:
public class DefaultClientConfigImpl implements IClientConfig { // ping 的默认策略 DummyPing public static final String DEFAULT_NFLOADBALANCER_PING_CLASSNAME = "com.netflix.loadbalancer.DummyPing"; // DummyPing.class.getName(); public static final String DEFAULT_NFLOADBALANCER_RULE_CLASSNAME = "com.netflix.loadbalancer.AvailabilityFilteringRule"; public static final String DEFAULT_NFLOADBALANCER_CLASSNAME = "com.netflix.loadbalancer.ZoneAwareLoadBalancer"; public static final int DEFAULT_MAX_TOTAL_TIME_TO_PRIME_CONNECTIONS = 30000; public static final int DEFAULT_MAX_RETRIES_PER_SERVER_PRIME_CONNECTION = 9; ............................................. }
IRule 表示 负载均衡策略,即如何去选择服务实例,默认为 RoundRobinRule,即通过轮询的方式选择服务。Ribbon 默认提供的有 7 种。
IPing 表示检测服务是否可用策略,Ribbon 也提供了很多策略,共有 5 种,默认为 DummyPing
关于 IRule 和 IPing 的策略,后面会专门进行研究。
在 BaseLoadBalancer 中,除了提供一个无参的构造方法(使用的是默认的配置)外,还提供了很多重载的构造方法,下面来看下根据客户端的配置来创建BaseLoadBalancer :
// 根据客户端配置来创建 BaseLoadBalancer public BaseLoadBalancer(IClientConfig config) { initWithNiwsConfig(config); } @Override public void initWithNiwsConfig(IClientConfig clientConfig) { // 负载均衡策略 String ruleClassName = (String) clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerRuleClassName); // ping 策略 String pingClassName = (String) clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerPingClassName); IRule rule = (IRule) ClientFactory.instantiateInstanceWithClientConfig(ruleClassName, clientConfig); IPing ping = (IPing) ClientFactory.instantiateInstanceWithClientConfig(pingClassName, clientConfig); // 状态 LoadBalancerStats stats = createLoadBalancerStatsFromConfig(clientConfig); // 初始化配置 initWithConfig(clientConfig, rule, ping, stats); } void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping, LoadBalancerStats stats) { this.config = clientConfig; String clientName = clientConfig.getClientName(); this.name = clientName; // ping 的周期 int pingIntervalTime = Integer.parseInt(clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerPingInterval,Integer.parseInt("30"))); // 最大 ping 的次数 int maxTotalPingTime = Integer.parseInt(clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerMaxTotalPingTime,Integer.parseInt("2"))); setPingInterval(pingIntervalTime); setMaxTotalPingTime(maxTotalPingTime); setRule(rule); setPing(ping); setLoadBalancerStats(stats); rule.setLoadBalancer(this); if (ping instanceof AbstractLoadBalancerPing) { ((AbstractLoadBalancerPing) ping).setLoadBalancer(this); } ................. // 注册监控/可忽略 init(); }
在上面的构造方法中,可用根据客户端配置的信息来创建一个BaseLoadBalancer,如客户端可以配置负载均衡策略,ping的策略,ping的时间间隔和最大次数等。
在 Ribbon 中,负载均衡器多久才去更新获取服务列表呢?在 BaseLoadBalancer 类中,有一个 setupPingTask 方法,在该方法内部,会创建 PingTask 定时任务去检测服务的可用性,而 PingTask 又会创建 Pinger 对象,在 Pinger 对象的 runPinger() 方法中,会根据ping策略即 pingerStrategy 的 pingServers(ping, allServer) 来获取服务的可用性,主要逻辑如下:
void setupPingTask() { if (canSkipPing()) { return; } // 如果已经有了定时任务,则取消 if (lbTimer != null) { lbTimer.cancel(); } // 第二个参数为true,表示它是一个deamon线程 lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name, true); // 创建 PingTask, 它继承于 TimerTask,定时执行 run 方法 lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000); ...... } class PingTask extends TimerTask { public void run() { // 默认 pingStrategy = new SerialPingStrategy() new Pinger(pingStrategy).runPinger(); } } public void runPinger() throws Exception { // 如果正在ping,则返回 if (!pingInProgress.compareAndSet(false, true)) { return; // Ping in progress - nothing to do } // 所有的服务,包括不可用的服务 Server[] allServers = null; boolean[] results = null; Lock allLock = null; Lock upLock = null; try { allLock = allServerLock.readLock(); allLock.lock(); allServers = allServerList.toArray(new Server[allServerList.size()]); allLock.unlock(); // 所有服务的数量 int numCandidates = allServers.length; // 所有服务ping的结果 results = pingerStrategy.pingServers(ping, allServers); // 状态可用的服务列表 final List<Server> newUpList = new ArrayList<Server>(); // 状态改变的服务列表 final List<Server> changedServers = new ArrayList<Server>(); for (int i = 0; i < numCandidates; i++) { // 最新的状态 boolean isAlive = results[i]; Server svr = allServers[i]; // 老的状态 boolean oldIsAlive = svr.isAlive(); // 更新状态 svr.setAlive(isAlive); // 如果状态改变了,则放到集合中,会进行重新拉取 if (oldIsAlive != isAlive) { changedServers.add(svr); } // 状态可用的服务 if (isAlive) { newUpList.add(svr); } } upLock = upServerLock.writeLock(); upLock.lock(); upServerList = newUpList; upLock.unlock(); // 变态改变监听器 notifyServerStatusChangeListener(changedServers); } finally { // ping 完成 pingInProgress.set(false); } } // 检测服务的状态 @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; if (ping != null) { results[i] = ping.isAlive(servers[i]); } } return results; }
在上面的逻辑中,Ribbon 每10秒向 EurekaClient 发送 ping 来判断服务的可用性,如果服务的可用性发生了改变或服务的数量和之前的不一致,则会更新或重新拉取服务。有了这些服务之后,会根据负载均衡策略 IRule 来选择一个可用的服务。
在前文说到 Ribbon 客户端 RibbonLoadBalancerClient 选择服务的时候,最终会调用 ILoadBalancer.chooseServer 来选择服务,接下来就来看下这个方法:
public Server chooseServer(Object key) { ....... //rule= new RoundRobinRule() return rule.choose(key); .... }
关于 Ribbon 的负载均衡策略 IRule, Ribbon 提供了 7 种,后面再来分析,现在只需要知道通过 IRule 来选择服务就可以了。
在上面的分析中,Ribbon 会每10秒定时的去检测服务的可用性,如果服务状态发生了变化则重新获取,之后,再根据负载均衡策略 IRule 来选择一个可用的服务;但是,在初始化的时候,是从哪里获取服务列表呢?下面就来分析这个问题
BaseLoadBalancer 有个子类 DynamicServerListLoadBalancer,它具有使用动态源获取服务器列表的功能。即服务器列表在运行时可能会更改。此外,还可以通过条件来过滤掉不符合所需条件的服务。
public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer { // 是否正在进行服务列表的更新 protected AtomicBoolean serverListUpdateInProgress = new AtomicBoolean(false); // 服务列表 volatile ServerList<T> serverListImpl; // 服务过滤器 volatile ServerListFilter<T> filter; }
在 DynamicServerListLoadBalancer 中,有个 restOfInit 方法,在初始化时进行调用,在该方法中,会从 Eureka 客户端中拉取所有的服务列表:
void restOfInit(IClientConfig clientConfig) { ............. updateListOfServers(); ........ } public void updateListOfServers() { List<T> servers = new ArrayList<T>(); if (serverListImpl != null) { // 获取所有服务列表 servers = serverListImpl.getUpdatedListOfServers(); // 根据条件过滤服务 if (filter != null) { servers = filter.getFilteredListOfServers(servers); } } updateAllServerList(servers); } protected void updateAllServerList(List<T> ls) { if (serverListUpdateInProgress.compareAndSet(false, true)) { try { for (T s : ls) { s.setAlive(true); // 状态设置为可用 } setServersList(ls); super.forceQuickPing(); // 强制检测服务状态 } finally { serverListUpdateInProgress.set(false); } } }
获取所有服务列表 servers = serverListImpl.getUpdatedListOfServers(); 最终会调用 DiscoveryEnabledNIWSServerList 的方法:
servers = serverListImpl.getUpdatedListOfServers(); public List<DiscoveryEnabledServer> getUpdatedListOfServers(){ return obtainServersViaDiscovery(); } private List<DiscoveryEnabledServer> obtainServersViaDiscovery() { List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>(); ........ // 通过 eurekaClient 来获取注册的服务列表 EurekaClient eurekaClient = eurekaClientProvider.get(); if (vipAddresses!=null){ for (String vipAddress : vipAddresses.split(",")) { List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion); for (InstanceInfo ii : listOfInstanceInfo) { if (ii.getStatus().equals(InstanceStatus.UP)) { ..... DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr); serverList.add(des); } } ...... } } return serverList; }
通过上面方法的分析,Ribbon 最终会通过 EurekaClient 来获取服务列表的,而 EurekaClient 的实现类是 DiscoveryClient,而在 Eureka 中,DiscoveryClient 类具有服务的注册,发现,续约,获取服务列表等功能。
此外,该类中还可以通过过滤器来获取不符合条件的服务。
以上就是 Ribbon 负载均衡器的一个实现原理。最后再来看下流程图,加深印象:
关于Ribbon中怎么使用 LoadBalancer 实现负载均衡就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。