您现在的位置是:首页 > 文章详情

SpringCloud源码:Ribbon负载均衡分析

日期:2018-12-09点击:376

    本文主要分析 SpringCloud 中 Ribbon 负载均衡流程和原理。

    SpringCloud版本为:Edgware.RELEASE。

一.时序图

    和以前一样,先把图贴出来,直观一点:

 

二.源码分析

    我们先从 contoller 里面看如何使用 Ribbon 来负载均衡的:

 @GetMapping("/user/{id}") public User findById(@PathVariable Long id) { //return this.restTemplate.getForObject("http://192.168.2.110:8001/" + id, User.class); return this.restTemplate.getForObject("http://microservice-provider-user/" + id, User.class); }

    可以看到,在整合 Ribbon 之前,请求Rest是通过IP端口直接请求。整合 Ribbon 之后,请求的地址改成了 http://applicationName ,官方取名为虚拟主机名(virtual host name),当 Ribbon 和 Eureka 配合使用时,会自动将虚拟主机名转换为微服务的实际IP地址,我们后面会分析这个过程。

    首先从 RestTemplate#getForObject 开始:

 public <T> T getForObject(String url, Class<T> responseType, Object... uriVariables) throws RestClientException { // 设置RequestCallback的返回类型responseType RequestCallback requestCallback = acceptHeaderRequestCallback(responseType); // 实例化responseExtractor HttpMessageConverterExtractor<T> responseExtractor = new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger); return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables); }

    接着执行到 RestTemplate 的 execute,主要是拼装URI,如果存在baseUrl,则插入baseUrl。拼装好后,进入实际"执行"请求的地方:

 public <T> T execute(String url, HttpMethod method, RequestCallback requestCallback, ResponseExtractor<T> responseExtractor, Object... uriVariables) throws RestClientException { // 组装 URI URI expanded = getUriTemplateHandler().expand(url, uriVariables); // 实际"执行"的地方 return doExecute(expanded, method, requestCallback, responseExtractor); }

    RestTemplate#doExecute,实际“执行”请求的地方,执行超过后,返回 response:

 protected <T> T doExecute(URI url, HttpMethod method, RequestCallback requestCallback, ResponseExtractor<T> responseExtractor) throws RestClientException { ClientHttpResponse response = null; try { // 实例化请求,url为请求地址,method为GET ClientHttpRequest request = createRequest(url, method); if (requestCallback != null) {// AcceptHeaderRequestCallback requestCallback.doWithRequest(request); } // 实际处理请求的地方 response = request.execute(); // 处理response,记录日志和调用对应的错误处理器 handleResponse(url, method, response); if (responseExtractor != null) {// 使用前面的HttpMessageConverterExtractor从Response里面抽取数据 return responseExtractor.extractData(response); } else { return null; } } ...... }

    到了请求被执行的地方,AbstractClientHttpRequest#execute,跳转到 executeInternal:

 public final ClientHttpResponse execute() throws IOException { // 断言请求还没被执行过 assertNotExecuted(); // 跳转到 executeInternal 处理请求 ClientHttpResponse result = executeInternal(this.headers); // 标记请求为已经执行过 this.executed = true; return result; }

    AbstractBufferingClientHttpRequest#executeInternal,AbstractBufferingClientHttpRequest是AbstractClientHttpRequest的子抽象类,作用是缓存output,使用了一个字节数组输出流:

 protected ClientHttpResponse executeInternal(HttpHeaders headers) throws IOException { // 首次进来,bytes内容为空 byte[] bytes = this.bufferedOutput.toByteArray(); if (headers.getContentLength() < 0) { // 设置 Content-Length 为 1 headers.setContentLength(bytes.length); } // 模板方法,跳转到了实现类中的方法,InterceptingClientHttpRequest#executeInternal ClientHttpResponse result = executeInternal(headers, bytes); // 拿到结果后,清空缓存 this.bufferedOutput = null; return result; }

    executeInternal是一个抽象方法,跳转到了其实现类InterceptingClientHttpRequest#executeInternal:

 protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException { InterceptingRequestExecution requestExecution = new InterceptingRequestExecution(); // InterceptingRequestExecution是一个内部类 return requestExecution.execute(this, bufferedOutput); } // 内部类,负责执行请求 private class InterceptingRequestExecution implements ClientHttpRequestExecution { private final Iterator<ClientHttpRequestInterceptor> iterator;// 所有HttpRequest的拦截器 public InterceptingRequestExecution() { this.iterator = interceptors.iterator(); } @Override public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException { if (this.iterator.hasNext()) {// 如果还有下一个拦截器,则执行其拦截方法 // 这里的拦截器是 MetricsClientHttpRequestInterceptor,对应"metrics"信息,记录执行时间和结果 ClientHttpRequestInterceptor nextInterceptor = this.iterator.next(); // 执行拦截方法 return nextInterceptor.intercept(request, body, this); } ...... } }

    跳转到了拦截器 MetricsClientHttpRequestInterceptor 的拦截方法:

 public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException { long startTime = System.nanoTime();// 标记开始执行时间 ClientHttpResponse response = null; try { // 传入请求和Body,处理执行,又跳转回 InterceptingRequestExecution response = execution.execute(request, body); return response; } finally {// 在执行完方法,返回response之前,记录一下执行的信息 SmallTagMap.Builder builder = SmallTagMap.builder(); for (MetricsTagProvider tagProvider : tagProviders) { for (Map.Entry<String, String> tag : tagProvider .clientHttpRequestTags(request, response).entrySet()) { builder.add(Tags.newTag(tag.getKey(), tag.getValue())); } } MonitorConfig.Builder monitorConfigBuilder = MonitorConfig .builder(metricName); monitorConfigBuilder.withTags(builder); // 记录执行时间 servoMonitorCache.getTimer(monitorConfigBuilder.build()) .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); } }

    又跳转回了 InterceptingRequestExecution,下个拦截器是 - LoadBalancerInterceptor,最后的Boss,调用LoadBalancerClient完成请求的负载。

    LoadBalancerInterceptor#intercept,主角登场了,终于等到你,还好没放弃:

 public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException { // 获取原始URI final URI originalUri = request.getURI(); // 获取请求中的服务名字,也就是所谓的"虚拟主机名" String serviceName = originalUri.getHost(); // 转由 LoadBalancerClient 处理请求 return this.loadBalancer.execute(serviceName, requestFactory.createRequest(request, body, execution)); }

    下面空一行先停下来休息一下,然后看看,负载均衡是怎样实现的。

 

    LoadBalancerInterceptor这里默认的实现是 RibbonLoadBalancerClient,Ribbon是Netflix发布的负载均衡器。

    RibbonLoadBalancerClient#execute,负载均衡算法选出实际处理请求的Server:

 public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException { // serviceId即前面的虚拟主机名 "microservice-provider-user",获取loadBalancer //这里获取到的是 DynamicServerListLoadBalancer ILoadBalancer loadBalancer = getLoadBalancer(serviceId); // 基于loadBalancer,选择实际处理请求的服务提供者 Server server = getServer(loadBalancer); 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); }

    RibbonLoadBalancerClient#getServer,转交 loadBalancer 选择Server:

 protected Server getServer(ILoadBalancer loadBalancer) { if (loadBalancer == null) { return null; } // 由 loadBalancer 完成选Server的重任,这里的 key 是默认值 "default" return loadBalancer.chooseServer("default"); // TODO: better handling of key }

    chooseServer也是一个抽象的模板方法,最后的实现是 ZoneAwareLoadBalancer#chooseServer:

 public Server chooseServer(Object key) { if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) { logger.debug("Zone aware logic disabled or there is only one zone"); // 到了 BaseLoadBalancer的chooseServer return super.chooseServer(key); } ...... }

    BaseLoadBalancer#chooseServer,转交规则来选择Server:

 public Server chooseServer(Object key) { if (counter == null) { counter = createCounter(); } // counter是一个计数器,起始值是"0",下面自增一次,变为 "1" counter.increment(); if (rule == null) { return null; } else { try { // 默认的挑选规则是 "ZoneAvoidanceRule" return rule.choose(key); } catch (Exception e) { logger.warn("LoadBalancer [{}]: Error choosing server for key {}", name, key, e); return null; } } }

    PredicateBasedRule是ZoneAvoidanceRule的父类。PredicateBasedRule#choose,可以看到,基础负载规则采用的是"RoundRobin"即轮询的方式:

 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; } }

    下面分析"轮询"的过程,AbstractServerPredicate#chooseRoundRobinAfterFiltering,传入Server列表的长度,自增取模实现:

 public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) { // 首先拿到所有"合格"的Server List<Server> eligible = getEligibleServers(servers, loadBalancerKey); if (eligible.size() == 0) { return Optional.absent(); } // 在 incrementAndGetModulo 中获取,"自增取模" return Optional.of(eligible.get(incrementAndGetModulo(eligible.size()))); }

    AbstractServerPredicate#incrementAndGetModulo,维护了一个nextIndex,记录下次请求的下标:

 private int incrementAndGetModulo(int modulo) { for (;;) { int current = nextIndex.get();// 第一次 current是"0" int next = (current + 1) % modulo;// current+1对size取模,作为下次的"current" // "0" == current,则以原子方式将该值设置为 next if (nextIndex.compareAndSet(current, next)) return current; } }

    最后,我们通过控制台来验证一下请求是不是"轮询"分配到服务提供者的,本地启动了8000和8001两个Provider:

2018-12-09 18:55:30.794 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8001 2018-12-09 18:55:33.196 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8000 2018-12-09 18:55:34.713 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8001 2018-12-09 18:55:34.975 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8000 2018-12-09 18:55:35.175 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8001 2018-12-09 18:55:35.351 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8000 2018-12-09 18:55:35.534 c.i.c.s.user.controller.MovieController : microservice-provider-user:192.168.2.117:8001

    可以看到,请求确实被轮询给两个Provider处理的。

 

    至此,我们完成了 SpringCloud 中 Ribbon 负载均衡的过程,知道了默认采用的是"轮询"的方式,实现是通过维护一个index,自增后取模来作为下标挑选实际响应请求的Server。除了轮询的方式,还有随机等算法。感兴趣可以按照类似思路分析测试一下。

原文链接:https://my.oschina.net/javamaster/blog/2985895
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章