【一起学源码-微服务】Feign 源码三：Feign结合Ribbon实现负载均衡的原理分析
前言
前情回顾
上一讲我们已经知道了Feign的工作原理其实是在项目启动的时候，通过JDK动态代理为每个FeignClinent生成一个动态代理。
动态代理的数据结构是：ReflectiveFeign.FeignInvocationHandler。其中包含 target (里面是serviceName等信息)和 dispatcher (map数据结构，key是请求的方法名，方法参数等，value是 SynchronousMethodHandler )。
如下图所示：
本讲目录
这一讲主要是Feign与Ribbon结合实现负载均衡的原理分析。
说明
原创不易，如若转载 请标明来源！
博客地址：一枝花算不算浪漫
微信公众号：壹枝花算不算浪漫
源码分析
Feign结合Ribbon实现负载均衡原理
通过前面的分析，我们可以直接来看下 SynchronousMethodHandler 中的代码：
final class SynchronousMethodHandler implements MethodHandler {
@Override
public Object invoke(Object[] argv) throws Throwable {
// 生成请求类似于：GET /sayHello/wangmeng HTTP/1.1
RequestTemplate template = buildTemplateFromArgs.create(argv);
Retryer retryer = this.retryer.clone();
while (true) {
try {
return executeAndDecode(template);
} catch (RetryableException e) {
retryer.continueOrPropagate(e);
if (logLevel != Logger.Level.NONE) {
logger.logRetry(metadata.configKey(), logLevel);
}
continue;
}
}
}
Object executeAndDecode(RequestTemplate template) throws Throwable {
// 构建request对象：GET http://serviceA/sayHello/wangmeng HTTP/1.1
Request request = targetRequest(template);
if (logLevel != Logger.Level.NONE) {
logger.logRequest(metadata.configKey(), logLevel, request);
}
Response response;
long start = System.nanoTime();
try {
// 这个client就是之前构建的LoadBalancerFeignClient，options是超时时间
response = client.execute(request, options);
// ensure the request is set. TODO: remove in Feign 10
response.toBuilder().request(request).build();
} catch (IOException e) {
if (logLevel != Logger.Level.NONE) {
logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime(start));
}
throw errorExecuting(request, e);
}
long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
// 下面逻辑都是构建返回值response
boolean shouldClose = true;
try {
if (logLevel != Logger.Level.NONE) {
response =
logger.logAndRebufferResponse(metadata.configKey(), logLevel, response, elapsedTime);
// ensure the request is set. TODO: remove in Feign 10
response.toBuilder().request(request).build();
}
if (Response.class == metadata.returnType()) {
if (response.body() == null) {
return response;
}
if (response.body().length() == null ||
response.body().length() > MAX_RESPONSE_BUFFER_SIZE) {
shouldClose = false;
return response;
}
// Ensure the response body is disconnected
byte[] bodyData = Util.toByteArray(response.body().asInputStream());
return response.toBuilder().body(bodyData).build();
}
if (response.status() >= 200 && response.status() < 300) {
if (void.class == metadata.returnType()) {
return null;
} else {
return decode(response);
}
} else if (decode404 && response.status() == 404 && void.class != metadata.returnType()) {
return decode(response);
} else {
throw errorDecoder.decode(metadata.configKey(), response);
}
} catch (IOException e) {
if (logLevel != Logger.Level.NONE) {
logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime);
}
throw errorReading(request, response, e);
} finally {
if (shouldClose) {
ensureClosed(response.body());
}
}
}
}
这里主要是构建request数据，然后通过request和options去通过 LoadBalancerFeignClient.execute() 方法去获得返回值。我们可以接着看client端的调用：
public class LoadBalancerFeignClient implements Client {
@Override
public Response execute(Request request, Request.Options options) throws IOException {
try {
// asUri: http://serviceA/sayHello/wangmeng
URI asUri = URI.create(request.url());
// clientName:serviceA
String clientName = asUri.getHost();
// uriWithoutHost: http://sayHello/wangmeng
URI uriWithoutHost = cleanUrl(request.url(), clientName);
// 这里ribbonRequest：GET http:///sayHello/wangmeng HTTP/1.1
FeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(
this.delegate, request, uriWithoutHost);
// 这里面config只有两个超时时间，一个是connectTimeout：5000，一个是readTimeout：5000
IClientConfig requestConfig = getClientConfig(options, clientName);
// 真正执行负载均衡的地方
return lbClient(clientName).executeWithLoadBalancer(ribbonRequest,
requestConfig).toResponse();
}
catch (ClientException e) {
IOException io = findIOException(e);
if (io != null) {
throw io;
}
throw new RuntimeException(e);
}
}
}
接着我们看下 lbClient() 和 executeWithLoadBalancer()
public class LoadBalancerFeignClient implements Client {
private FeignLoadBalancer lbClient(String clientName) {
return this.lbClientFactory.create(clientName);
}
}
public class CachingSpringLoadBalancerFactory {
public FeignLoadBalancer create(String clientName) {
if (this.cache.containsKey(clientName)) {
return this.cache.get(clientName);
}
IClientConfig config = this.factory.getClientConfig(clientName);
// 获取Ribbon ILoadBalancer信息
ILoadBalancer lb = this.factory.getLoadBalancer(clientName);
ServerIntrospector serverIntrospector = this.factory.getInstance(clientName, ServerIntrospector.class);
FeignLoadBalancer client = enableRetry ? new RetryableFeignLoadBalancer(lb, config, serverIntrospector,
loadBalancedRetryPolicyFactory, loadBalancedBackOffPolicyFactory, loadBalancedRetryListenerFactory) : new FeignLoadBalancer(lb, config, serverIntrospector);
this.cache.put(clientName, client);
return client;
}
}
这里是获取了 ILoadBalancer 数据，里面包含了Ribbon获取的serviceA所有服务节点信息。
这里已经获取到 ILoadBalancer ，里面包含serviceA服务器所有节点请求host信息。接下来就是从中负载均衡选择一个节点信息host出来。
public abstract class AbstractLoadBalancerAwareClient<S extends ClientRequest, T extends IResponse> extends LoadBalancerContext implements IClient<S, T>, IClientConfigAware {
public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {
LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);
try {
return command.submit(
new ServerOperation<T>() {
@Override
public Observable<T> call(Server server) {
URI finalUri = reconstructURIWithServer(server, request.getUri());
S requestForServer = (S) request.replaceUri(finalUri);
try {
return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));
}
catch (Exception e) {
return Observable.error(e);
}
}
})
.toBlocking()
.single();
} catch (Exception e) {
Throwable t = e.getCause();
if (t instanceof ClientException) {
throw (ClientException) t;
} else {
throw new ClientException(e);
}
}
}
}
public class LoadBalancerCommand<T> {
public Observable<T> submit(final ServerOperation<T> operation) {
final ExecutionInfoContext context = new ExecutionInfoContext();
if (listenerInvoker != null) {
try {
listenerInvoker.onExecutionStart();
} catch (AbortExecutionException e) {
return Observable.error(e);
}
}
final int maxRetrysSame = retryHandler.getMaxRetriesOnSameServer();
final int maxRetrysNext = retryHandler.getMaxRetriesOnNextServer();
// Use the load balancer
Observable<T> o =
(server == null ? selectServer() : Observable.just(server))
.concatMap(new Func1<Server, Observable<T>>() {
}
// 省略代码...
// selectServer是真正执行负载均衡的逻辑
private Observable<Server> selectServer() {
return Observable.create(new OnSubscribe<Server>() {
@Override
public void call(Subscriber<? super Server> next) {
try {
// loadBalancerURI是http:///sayHello/wangmeng， loadBalancerKey为null
Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey);
next.onNext(server);
next.onCompleted();
} catch (Exception e) {
next.onError(e);
}
}
});
}
}
public class LoadBalancerContext implements IClientConfigAware {
public Server getServerFromLoadBalancer(@Nullable URI original, @Nullable Object loadBalancerKey) throws ClientException {
String host = null;
int port = -1;
if (original != null) {
host = original.getHost();
}
if (original != null) {
Pair<String, Integer> schemeAndPort = deriveSchemeAndPortFromPartialUri(original);
port = schemeAndPort.second();
}
// 获取到ILoadBalancer，这里面有IRule的信息及服务节点所有信息
ILoadBalancer lb = getLoadBalancer();
if (host == null) {
// Partial URI or no URI Case
// well we have to just get the right instances from lb - or we fall back
if (lb != null){
// 这里就执行真正的chooseServer的逻辑了。默认的rule为ZoneAvoidanceZule
Server svc = lb.chooseServer(loadBalancerKey);
if (svc == null){
throw new ClientException(ClientException.ErrorType.GENERAL,
"Load balancer does not have available server for client: "
+ clientName);
}
host = svc.getHost();
if (host == null){
throw new ClientException(ClientException.ErrorType.GENERAL,
"Invalid Server for :" + svc);
}
logger.debug("{} using LB returned Server: {} for request {}", new Object[]{clientName, svc, original});
return svc;
}
// 省略代码
}
}
}
上面代码已经很清晰了，这里就是真正的通过ribbon的 rule.chooseServer() 负载均衡地选择了一个服务节点调用，debug如下：
到了这里feign与ribbon的分析也就结束了，返回请求url信息，然后得到response结果：
总结
上面已经分析了Feign与Ribbon的整合，最终还是落到Ribbon中的ILoadBalancer中，使用最后使用IRule去选择对应的server数据。
下一讲 会画一个很大的图，包含Feign、Ribbon、Eureka关联的图，里面会画出每个组件的细节及依赖关系。也算是学习至今的一个总复习了。
申明
本文章首发自本人博客： https://www.cnblogs.com/wang-meng 和公众号： 壹枝花算不算浪漫 ，如若转载请标明来源!
感兴趣的小伙伴可关注个人公众号：壹枝花算不算浪漫
正文到此结束
- 本文标签： 目录 NSA value client cache retry 负载均衡 IDE key build Service Feign https 数据 Select 微信公众号 ribbon struct UI HTML 总结 文章 原理分析 db map CTO final ACE bug 源码 cat CEO 参数 博客 id 服务器 find 时间 Eureka http 工作原理 stream spring tar 代码 list src IO 微服务
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
