spring cloud kubernetes源码解析之feign与loadbalancer
Posted 水中加点糖
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了spring cloud kubernetes源码解析之feign与loadbalancer相关的知识，希望对你有一定的参考价值。
现在是2021年8月的一个星期四，
窗外是刚下过大雨的夜色，伴着几点稀零的灯光。
我戴着耳机，
在客厅阳台上的书桌前吹着安静的风，
耳机中正好播放“夜风凛凛 独回望旧事前尘”张国荣的歌。
不知道不觉间，我来公司也已经二年了。
二年间匆匆过去，同样也预示着我也从大学毕业正好四年了。
在公司这二年间，我见证了公司的微服务从docker->docker swarm->kubernetes迁移的一路历程。
同时变化的也还有青涩的自己，从中级JAVA->项目技术负责人->team leader，一切感觉都是那么的真切。
一次次新模块产品的上线、一次次的重构后性能的极速提升、每次经历上万QPS并发时激动的心情，那些个激情澎湃的每一天。
直到一个多月前的一天，我接受了同学的建议，让我出去面试看看，试试外面更大规模公司的机会，几轮面试一路通过。
拿到offer后犹豫了很久，在同学的劝说下，我最终还是向公司提出了离职。
最后交接的两周，看着曾经写下的代码，重构后的部分模块，我竟后悔就不该一时冲动提出离职，也曾好几夜的胡思乱想与失眠。
尽管后面找了机会约了公司领导单独聊了聊，直到领导祝我在新的公司好好干后，我才明白职场中的忠诚度是什么。离职需谨慎！
既然没有了后悔的机会，也没有别的退路，那就勇往直前地向前冲吧(至少钱多)。
去新的公司好好干。
一夜过去，周五了。
我也早已想好，离职最后一天的我只需要在工位上静静地坐着，等着时间滴答滴答地走。一到点我起身，给周围同事一个笑容，之后默默地背起装着陪伴了我2年的办公用品的书包，给领导打声照顾说声再见。
再之后便是转身，只需走向办公室门外的电梯口，不再回头，按下电梯。
我看着电梯的门缓缓地合闭并关上。那一刻，也就是我和这可爱的公司、可爱的同事、可爱的领导短暂离别的那一刻。
再见了，约苗！
根据了解，新的公司项目中没有使用spring cloud相关的技术栈，去了新公司后也要和这个高大上的框架做一个短暂的告别，甚是不舍。
趁着离职的最后一天也没什么工作安排，在公司里最后再记录一下关于spring cloud kubernetes框架中关于loadbalancer部分源码的阅读笔记。
前言
结合公司产品的部署环境都是部署在k8s下，为了让微服务更加配合k8s的环境，在我的建议下，领导最终同意了采用spring cloud kubernetes框架作为spring cloud服务发现和治理的组件。
修改为spring cloud kubernetes之后的项目，在经过了1年多的线上运行，经历了上千万笔订单和在上万QPS的验证下，目前一切运行良好。
其中项目中服务调用服务是通过feign调用的，在feign调用的过程中会使用到loadbalancer，本文将通过阅读源码，来看一下在spring cloud kubernetes中feign调用时的核心代码链，以及spring cloud kubernetes中loadbacner两种负载模式获取最终url的过程与区别是什么。
spring cloud feign调用主要过程
feign.ReflectiveFeign.FeignInvocationHandler#invoke
feign.SynchronousMethodHandler#invoke
Object executeAndDecode(RequestTemplate template, Options options) throws Throwable {
Request request = targetRequest(template);
if (logLevel != Logger.Level.NONE) {
logger.logRequest(metadata.configKey(), logLevel, request);
}
Response response;
long start = System.nanoTime();
try {
response = client.execute(request, options);
org.springframework.cloud.openfeign.loadbalancer.RetryableFeignBlockingLoadBalancerClient#execute
@Override
public Response execute(Request request, Request.Options options) throws IOException {
final URI originalUri = URI.create(request.url());
String serviceId = originalUri.getHost();
Assert.state(serviceId != null, "Request URI does not contain a valid hostname: " + originalUri);
final LoadBalancedRetryPolicy retryPolicy = loadBalancedRetryFactory.createRetryPolicy(serviceId,
loadBalancerClient);
RetryTemplate retryTemplate = buildRetryTemplate(serviceId, request, retryPolicy);
return retryTemplate.execute(context -> {
Request feignRequest = null;
ServiceInstance retrievedServiceInstance = null;
Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(
loadBalancerClientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
RetryableRequestContext.class, ResponseData.class, ServiceInstance.class);
String hint = getHint(serviceId);
DefaultRequest<RetryableRequestContext> lbRequest = new DefaultRequest<>(
new RetryableRequestContext(null, buildRequestData(request), hint));
// On retries the policy will choose the server and set it in the context
// and extract the server and update the request being made
if (context instanceof LoadBalancedRetryContext) {
LoadBalancedRetryContext lbContext = (LoadBalancedRetryContext) context;
ServiceInstance serviceInstance = lbContext.getServiceInstance();
if (serviceInstance == null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Service instance retrieved from LoadBalancedRetryContext: was null. "
+ "Reattempting service instance selection");
}
ServiceInstance previousServiceInstance = lbContext.getPreviousServiceInstance();
lbRequest.getContext().setPreviousServiceInstance(previousServiceInstance);
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
//通过负载均衡器获取一个instance
retrievedServiceInstance = loadBalancerClient.choose(serviceId, lbRequest);
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Selected service instance: %s", retrievedServiceInstance));
}
lbContext.setServiceInstance(retrievedServiceInstance);
}
if (retrievedServiceInstance == null) {
if (LOG.isWarnEnabled()) {
LOG.warn("Service instance was not resolved, executing the original request");
}
org.springframework.cloud.client.loadbalancer.Response<ServiceInstance> lbResponse = new DefaultResponse(
retrievedServiceInstance);
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<ResponseData, ServiceInstance, RetryableRequestContext>(
CompletionContext.Status.DISCARD, lbRequest, lbResponse)));
feignRequest = request;
}
else {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Using service instance from LoadBalancedRetryContext: %s",
retrievedServiceInstance));
}
String reconstructedUrl = loadBalancerClient.reconstructURI(retrievedServiceInstance, originalUri)
.toString();
feignRequest = buildRequest(request, reconstructedUrl);
}
}
org.springframework.cloud.client.loadbalancer.Response<ServiceInstance> lbResponse = new DefaultResponse(
retrievedServiceInstance);
Response response = LoadBalancerUtils.executeWithLoadBalancerLifecycleProcessing(delegate, options,
feignRequest, lbRequest, lbResponse, supportedLifecycleProcessors,
retrievedServiceInstance != null);
int responseStatus = response.status();
if (retryPolicy != null && retryPolicy.retryableStatusCode(responseStatus)) {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Retrying on status code: %d", responseStatus));
}
response.close();
throw new RetryableStatusCodeException(serviceId, responseStatus, response, URI.create(request.url()));
}
return response;
}, new LoadBalancedRecoveryCallback<Response, Response>() {
@Override
protected Response createResponse(Response response, URI uri) {
return response;
}
});
}
org.springframework.cloud.loadbalancer.blocking.client.BlockingLoadBalancerClient#choose(java.lang.String, org.springframework.cloud.client.loadbalancer.Request)
@Override
public <T> ServiceInstance choose(String serviceId, Request<T> request) {
ReactiveLoadBalancer<ServiceInstance> loadBalancer = loadBalancerClientFactory.getInstance(serviceId);
if (loadBalancer == null) {
return null;
}
Response<ServiceInstance> loadBalancerResponse = Mono.from(loadBalancer.choose(request)).block();
if (loadBalancerResponse == null) {
return null;
}
return loadBalancerResponse.getServer();
}
org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer#choose
@SuppressWarnings("rawtypes")
@Override
// see original
// https://github.com/Netflix/ocelli/blob/master/ocelli-core/
// src/main/java/netflix/ocelli/loadbalancer/RoundRobinLoadBalancer.java
public Mono<Response<ServiceInstance>> choose(Request request) {
ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider
.getIfAvailable(NoopServiceInstanceListSupplier::new);
return supplier.get(request).next()
.map(serviceInstances -> processInstanceResponse(supplier, serviceInstances));
}
org.springframework.cloud.context.named.ClientFactoryObjectProvider#getIfAvailable(java.util.function.Supplier)
public T getIfAvailable(Supplier<T> defaultSupplier) throws BeansException {
return this.delegate().getIfAvailable(defaultSupplier);
}
org.springframework.cloud.context.named.ClientFactoryObjectProvider#getIfAvailable(java.util.function.Supplier)
@Override
public T getIfAvailable(Supplier<T> defaultSupplier) throws BeansException {
return delegate().getIfAvailable(defaultSupplier);
}
org.springframework.beans.factory.support.DefaultListableBeanFactory#getBeanProvider(org.springframework.core.ResolvableType, boolean)
@Override
public <T> ObjectProvider<T> getBeanProvider(ResolvableType requiredType, boolean allowEagerInit) {
//org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier
return new BeanObjectProvider<T>() {
@Override
public T getObject() throws BeansException {
T resolved = resolveBean(requiredType, null, false);
if (resolved == null) {
throw new NoSuchBeanDefinitionException(requiredType);
}
return resolved;
}
@Override
public T getObject(Object... args) throws BeansException {
T resolved = resolveBean(requiredType, args, false);
if (resolved == null) {
throw new NoSuchBeanDefinitionException(requiredType);
}
return resolved;
}
@Override
@Nullable
public T getIfAvailable() throws BeansException {
try {
return resolveBean(requiredType, null, false);
}
catch (ScopeNotActiveException ex) {
// Ignore resolved bean in non-active scope
return null;
}
}
@Override
public void ifAvailable(Consumer<T> dependencyConsumer) throws BeansException {
T dependency = getIfAvailable();
if (dependency != null) {
try {
dependencyConsumer.accept(dependency);
}
catch (ScopeNotActiveException ex) {
// Ignore resolved bean in non-active scope, even on scoped proxy invocation
}
}
}
@Override
@Nullable
public T getIfUnique() throws BeansException {
try {
return resolveBean(requiredType, null, true);
}
catch (ScopeNotActiveException ex) {
// Ignore resolved bean in non-active scope
return null;
}
}
@Override
public void ifUnique(Consumer<T> dependencyConsumer) throws BeansException {
T dependency = getIfUnique();
if (dependency != null) {
try {
dependencyConsumer.accept(dependency);
}
catch (ScopeNotActiveException ex) {
// Ignore resolved bean in non-active scope, even on scoped proxy invocation
}
}
}
@SuppressWarnings("unchecked")
@Override
public Stream<T> stream() {
return Arrays.stream(getBeanNamesForTypedStream(requiredType, allowEagerInit))
.map(name -> (T) getBean(name))
.filter(bean -> !(bean instanceof NullBean));
}
@SuppressWarnings("unchecked")
@Override
public Stream<T> orderedStream() {
String[] beanNames = getBeanNamesForTypedStream(requiredType, allowEagerInit);
if (beanNames.length == 0) {
return Stream.empty();
}
Map<String, T> matchingBeans = CollectionUtils.newLinkedHashMap(beanNames.length);
for (String beanName : beanNames) {
Object beanInstance = getBean(beanName);
if (!(beanInstance instanceof NullBean)) {
matchingBeans.put(beanName, (T) beanInstance);
}
}
Stream<T> stream = matchingBeans.values().stream();
return stream.sorted(adaptOrderComparator(matchingBeans));
}
};
}
org.springframework.beans.factory.support.DefaultListableBeanFactory.BeanObjectProvider#getIfAvailable会调用resolveBean,
org.springframework.beans.factory.support.DefaultListableBeanFactory#resolveBean
@Nullable
private <T> T resolveBean(ResolvableType requiredType, @Nullable Object[] args, boolean nonUniqueAsNull) {
NamedBeanHolder<T> namedBean = resolveNamedBean(requiredType, args, nonUniqueAsNull);
if (namedBean != null) {
return namedBean.getBeanInstance();
}
BeanFactory parent = getParentBeanFactory();
if (parent instanceof DefaultListableBeanFactory) {
return ((DefaultListableBeanFactory) parent).resolveBean(requiredType, args, nonUniqueAsNull);
}
else if (parent != null) {
ObjectProvider<T> parentProvider = parent.getBeanProvider(requiredType);
if (args != null) {
return parentProvider.getObject(args);
}
else {
return (nonUniqueAsNull ? parentProvider.getIfUnique() : parentProvider.getIfAvailable());
}
}
return null;
}
上面的resolveNamedBean方法：
org.springframework.beans.factory.support.DefaultListableBeanFactory#resolveNamedBean(org.springframework.core.ResolvableType, java.lang.Object[], boolean)
@Nullable
private <T> NamedBeanHolder<T> resolveNamedBean(
ResolvableType requiredType, @Nullable Object[] args, boolean nonUniqueAsNull) throws BeansException {
//org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier
Assert.notNull(requiredType, "Required type must not be null");
String[] candidateNames = getBeanNamesForType(requiredType);
if (candidateNames.length > 1) {
List<String> autowireCandidates = new ArrayList<>(candidateNames.length);
for (String beanName : candidateNames) {
if (!containsBeanDefinition(beanName) || getBeanDefinition(beanName).isAutowireCandidate()) {
autowireCandidates.add(beanName);
}
}
if (!autowireCandidates.isEmpty()) {
candidateNames = StringUtils.toStringArray(autowireCandidates);
}
}
if (candidateNames.length == 1) {
return resolveNamedBean(candidateNames[0], requiredType, args);
}
else if (candidateNames.length > 1) {
Map<String, Object> candidates = CollectionUtils.newLinkedHashMap(candidateNames.length);
for (String beanName : candidateNames) {
if (containsSingleton(beanName) && args == null) {
Object beanInstance = getBean(beanName);
candidates.put(beanName, (beanInstance instanceof NullBean ? null : beanInstance));
}
else {
candidates.put(beanName, getType(beanName));
}
}
String candidateName = determinePrimaryCandidate(candidates, requiredType.toClass());
if (candidateName == null) {
candidateName = determineHighestPriorityCandidate(candidates, requiredType.toClass());
}
if (candidateName != null) {
Object beanInstance = candidates.get(candidateName);
if (beanInstance == null) {
return null;
}
if (beanInstance instanceof Class) {
return resolveNamedBean(candidateName, requiredType, args);
}
return new NamedBeanHolder<>(candidateName, (T) beanInstance);
}
if (!nonUniqueAsNull) {
throw new NoUniqueBeanDefinitionException(requiredType, candidates.keySet());
}
}
return null;
}
org.springframework.beans.factory.support.DefaultListableBeanFactory#getBeanNamesForType(org.springframework.core.ResolvableType)
@Override
public String[] getBeanNamesForType(ResolvableType type) {
//org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier
return getBeanNamesForType(type, true, true);
}
@Override
public String[] getBeanNamesForType(ResolvableType type, boolean includeNonSingletons, boolean allowEagerInit) {
//org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier
Class<?> resolved = type.resolve();
if (resolved != null && !type.hasGenerics()) {
return getBeanNamesForType(resolved, includeNonSingletons, allowEagerInit);
}
else {
return doGetBeanNamesForType(type, includeNonSingletons, allowEagerInit);
}
}
最终获得：
service:
org.springframework.cloud.kubernetes.client.loadbalancer.KubernetesClientServicesListSupplier
pod:
org.springframework.cloud.loadbalancer.core.CachingServiceInstanceListSupplier
service模式下获取远程访问地址过程
待更新,此模式下默认不会走缓存
pod模式下获取远程访问地址过程
待更新,此模式下默认会走缓存
config
org.springframework.cloud.loadbalancer.config.LoadBalancerCacheAutoConfiguration
待更新，具体初始化过程
以上是关于spring cloud kubernetes源码解析之feign与loadbalancer的主要内容，如果未能解决你的问题，请参考以下文章
spring cloud kubernetes在pod模式下服务调用源码解析
spring cloud kubernetes在pod模式下服务调用源码解析
spring cloud kubernetes源码解析之feign与loadbalancer
spring cloud kubernetes源码解析之feign与loadbalancer