spring cloud kubernetes源码解析之feign与loadbalancer

柔情只为你懂 2022-09-14 03:04 813阅读 0赞

现在是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

待更新，具体初始化过程