springCloud微服务系列——链路跟踪第三篇——feign链路跟踪器

小咪咪 2022-05-18 10:34 394阅读 0赞

**目录**

一、简介

二、思路

三、获取riboon负载均衡结果

源码分析

扩展点

四、示例代码

--------------------

# 一、简介 #

这篇文章总结feign链路跟踪器的实现

# 二、思路 #

上篇文章中总结了mvc的链路跟踪器，我们可以知道要实现链路跟踪器需要在前后拦截请求，那么这里我们应该怎么处理呢？既然使用的是spring，那么很容易想到使用AOP来进行拦截。

还有个细节需要注意，因为feign可能会调用hystrix，这个时候hystrixCommand会新产生一个线程，这时候需要通过http将链路信息传递过去。参看[《springCloud微服务系列——OAuth2+JWT模式下的feign+hystrix处理》][springCloud_OAuth2_JWT_feign_hystrix]

@Slf4j
 public class LuminaryRequestInterceptor implements RequestInterceptor {
 
 	public static String TOKEN_HEADER = "authorization";
 	
 	/* (non-Javadoc) 
 	 * Title: apply 
 	 * Description: 
 	 * @param arg0 
 	 * @see feign.RequestInterceptor#apply(feign.RequestTemplate) 
 	 */
 	@Override
 	public void apply(RequestTemplate template) {
 		HttpServletRequest request = getHttpServletRequest();
 		if(request == null) {
 			log.warn("request is null in feign request interceptor");
 			return;
 		}
 		Map<String, String> headerMap = getHeaders(request);
 		if(headerMap.get(TOKEN_HEADER) != null) {
 			template.header(TOKEN_HEADER, headerMap.get(TOKEN_HEADER));
 		}
 		if(request.getAttribute(TraceInfo.TRACE_ID_KEY) != null) {
 			template.header(TraceInfo.TRACE_ID_KEY, (String) request.getAttribute(TraceInfo.TRACE_ID_KEY));
 		}
 		if(request.getAttribute(TraceInfo.RPC_ID_KEY) != null) {
 			template.header(TraceInfo.RPC_ID_KEY, (String) request.getAttribute(TraceInfo.RPC_ID_KEY));
 		}
 	}
 	
 	private HttpServletRequest getHttpServletRequest() {
 try {
 return ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
 } catch (Exception e) {
 return null;
 }
 }
 	
 	private Map<String, String> getHeaders(HttpServletRequest request) {
 Map<String, String> map = new LinkedHashMap<>();
 Enumeration<String> enumeration = request.getHeaderNames();
 while (enumeration.hasMoreElements()) {
 String key = enumeration.nextElement();
 String value = request.getHeader(key);
 map.put(key, value);
 }
 return map;
 }
 
 }

# 三、获取riboon负载均衡结果 #

feign链路跟踪的时候还有个获取负载均衡结果的问题，我们需要在跟踪的时候拿到具体调用的是哪个服务。feign的负载均衡使用的是riboon，在riboon中通过eureka拿到服务列表，通过ping的方式筛选出可用的服务。

##   源码分析 ##

![70][]

其中紫色表示类，绿色表示方法，蓝色表示内部方法

我们可以看到最终是注册成了一个观察者，在特定的事件中通过loadBalancerContext的getServerFromLoadBalancer方法，loadBalancerContext中又通过ILoadBalancer调用chooseServer方法，最终通过IRule调用choose获得负载均衡结果。

LoadBalancerFeignClient

@Override
    	public Response execute(Request request, Request.Options options) throws IOException {
    		try {
    			URI asUri = URI.create(request.url());
    			String clientName = asUri.getHost();
    			URI uriWithoutHost = cleanUrl(request.url(), clientName);
    			FeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(
    					this.delegate, request, uriWithoutHost);
    
    			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);
    		}
    	}

AbstractLoadBalancerAwareClient

public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {
 RequestSpecificRetryHandler handler = getRequestSpecificRetryHandler(request, requestConfig);
 LoadBalancerCommand<T> command = LoadBalancerCommand.<T>builder()
 .withLoadBalancerContext(this)
 .withRetryHandler(handler)
 .withLoadBalancerURI(request.getUri())
 .build();
 
 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);
 }
 }
 
 }

LoadBalancerCommand

private Observable<Server> selectServer() {
 return Observable.create(new OnSubscribe<Server>() {
 @Override
 public void call(Subscriber<? super Server> next) {
 try {
 Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey); 
 next.onNext(server);
 next.onCompleted();
 } catch (Exception e) {
 next.onError(e);
 }
 }
 });
 }

Observable

public static <T> Observable<T> create(OnSubscribe<T> f) {
 return new Observable<T>(RxJavaHooks.onCreate(f));
 }

RxJavaHooks

public static <T> Observable.OnSubscribe<T> onCreate(Observable.OnSubscribe<T> onSubscribe) {
 Func1<Observable.OnSubscribe, Observable.OnSubscribe> f = onObservableCreate;
 if (f != null) {
 return f.call(onSubscribe);
 }
 return onSubscribe;
 }

LoadBalancerContext

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();
 }
 
 // Various Supported Cases
 // The loadbalancer to use and the instances it has is based on how it was registered
 // In each of these cases, the client might come in using Full Url or Partial URL
 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){
 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;
 } else {
 // No Full URL - and we dont have a LoadBalancer registered to
 // obtain a server
 // if we have a vipAddress that came with the registration, we
 // can use that else we
 // bail out
 if (vipAddresses != null && vipAddresses.contains(",")) {
 throw new ClientException(
 ClientException.ErrorType.GENERAL,
 "Method is invoked for client " + clientName + " with partial URI of ("
 + original
 + ") with no load balancer configured."
 + " Also, there are multiple vipAddresses and hence no vip address can be chosen"
 + " to complete this partial uri");
 } else if (vipAddresses != null) {
 try {
 Pair<String,Integer> hostAndPort = deriveHostAndPortFromVipAddress(vipAddresses);
 host = hostAndPort.first();
 port = hostAndPort.second();
 } catch (URISyntaxException e) {
 throw new ClientException(
 ClientException.ErrorType.GENERAL,
 "Method is invoked for client " + clientName + " with partial URI of ("
 + original
 + ") with no load balancer configured. "
 + " Also, the configured/registered vipAddress is unparseable (to determine host and port)");
 }
 } else {
 throw new ClientException(
 ClientException.ErrorType.GENERAL,
 this.clientName
 + " has no LoadBalancer registered and passed in a partial URL request (with no host:port)."
 + " Also has no vipAddress registered");
 }
 }
 } else {
 // Full URL Case
 // This could either be a vipAddress or a hostAndPort or a real DNS
 // if vipAddress or hostAndPort, we just have to consult the loadbalancer
 // but if it does not return a server, we should just proceed anyways
 // and assume its a DNS
 // For restClients registered using a vipAddress AND executing a request
 // by passing in the full URL (including host and port), we should only
 // consult lb IFF the URL passed is registered as vipAddress in Discovery
 boolean shouldInterpretAsVip = false;
 
 if (lb != null) {
 shouldInterpretAsVip = isVipRecognized(original.getAuthority());
 }
 if (shouldInterpretAsVip) {
 Server svc = lb.chooseServer(loadBalancerKey);
 if (svc != null){
 host = svc.getHost();
 if (host == null){
 throw new ClientException(ClientException.ErrorType.GENERAL,
 "Invalid Server for :" + svc);
 }
 logger.debug("using LB returned Server: {} for request: {}", svc, original);
 return svc;
 } else {
 // just fall back as real DNS
 logger.debug("{}:{} assumed to be a valid VIP address or exists in the DNS", host, port);
 }
 } else {
 // consult LB to obtain vipAddress backed instance given full URL
 //Full URL execute request - where url!=vipAddress
 logger.debug("Using full URL passed in by caller (not using load balancer): {}", original);
 }
 }
 // end of creating final URL
 if (host == null){
 throw new ClientException(ClientException.ErrorType.GENERAL,"Request contains no HOST to talk to");
 }
 // just verify that at this point we have a full URL
 
 return new Server(host, port);
 }

BaseLoadBalancer

public String choose(Object key) {
            if (rule == null) {
                return null;
            } else {
                try {
                    Server svr = rule.choose(key);
                    return ((svr == null) ? null : svr.getId());
                } catch (Exception e) {
                    logger.warn("LoadBalancer [{}]:  Error choosing server", name, e);
                    return null;
                }
            }
        }

##   扩展点 ##

从源码来看，我们如果能获得LoadBalancerContext或者获得ILoaderBalancer都可以获得负载均衡的结果。

我们通过官方文档，发现可以自定义一些组件

![70 1][]

思路来了，我们可以自定义ILoadBalancer，但是这里其实有个坑，自定义的ILoadBalancer获得的server为null，我们再往上走一步，自定义IRule就可以了

最后我们需要实现监听者模式，因为我们并不知道负载均衡结果获取的时机，而我们要把该结果保存到链路信息中

# 四、示例代码 #

@Slf4j
 public class LuminaryRibbonRule extends ZoneAvoidanceRule implements ServerPublisher {
 
 	public static Queue<ServerListener> serverListeners;
 	
 	public static void regist(ServerListener serverListener) {
 		if(serverListeners == null)
 			 synchronized (LuminaryRibbonRule.class) {
 				 if(serverListeners == null) {
 					 serverListeners = new ConcurrentLinkedQueue<ServerListener>();
 				 }
 			 }
 		
 		serverListeners.add(serverListener);
 	}
 	
 	/* (non-Javadoc) 
 	 * Title: notifyAll 
 	 * Description: 
 	 * @param serverEvent 
 	 * @see com.luminary.component.ribbon.publisher.ServerPublisher#notifyAll(com.luminary.component.ribbon.event.ServerEvent) 
 	 */
 	@Override
 	public void notifyAll(ServerEvent serverEvent) {
 for(ServerListener serverListener : serverListeners) {
 	 serverListener.chooseServer(serverEvent);
 }
 	}
 	
 	@Override
 public Server choose(Object key) {
 		Server server = super.choose(key);
 if(server != null) {
 	log.info("select server:"+server.getHostPort());
 }
 ServerEvent serverEvent = new ServerEvent();
 serverEvent.setServer(server);
 notifyAll(serverEvent);
 return server;
 	}
 
 }

public interface ServerPublisher {
    
    	void notifyAll(ServerEvent serverEvent);
    	
    }

public interface ServerListener {
 
 	/**
 	 * 
 	 * Title: chooseServer 
 	 * Description: 选择服务事件监听逻辑 
 	 * @param serverEvent
 	 */
 	void chooseServer(ServerEvent serverEvent);
 	
 }

@Data
    public class ServerEvent {
    
    	private Server server;
    	
    }

@Slf4j
 @Aspect
 @Component
 public class FeignTracker extends GenericTracker implements Tracker<TraceHolder> {
 	
 	@Value("${spring.profiles.active:default}")
 	private String profile;
 	
 	@Autowired
 	private TraceClient traceClient;
 	
 	/** 
 	* Title: 
 	* Description: 
 	* @param traceClient 
 	*/ 
 	public FeignTracker(TraceClient traceClient) {
 		super(traceClient);
 		this.traceClient = traceClient;
 	}
 
 	@Around(value = "@annotation(trace)")
 	public Object proceed(ProceedingJoinPoint joinPoint, Trace trace) throws Throwable {
 		
 		if(!trace.value().getName().equals(this.getClass().getName())) {
 			return joinPoint.proceed();
 		}
 		
 		FeignTraceHolder traceHolder = new FeignTraceHolder();
 		Object result = null;
 		
 		try {
 			
 			log.info("feign tracker");
 			
 			Gson gson = new Gson();
 			
 			Signature signature = joinPoint.getSignature();
 		 MethodSignature methodSignature = (MethodSignature) signature;
 		 Method method = methodSignature.getMethod();
 		 String methodName = method.getName();
 
 		 FeignClient feignClient = AnnotationUtils.findAnnotation(method.getDeclaringClass(), FeignClient.class);
 		 
 		 GetMapping getMapping = method.getAnnotation(GetMapping.class);
 		 PostMapping postMapping = method.getAnnotation(PostMapping.class);
 		 PutMapping putMapping = method.getAnnotation(PutMapping.class);
 		 DeleteMapping deleteMapping = method.getAnnotation(DeleteMapping.class);
 		 RequestMapping requestMapping = method.getAnnotation(RequestMapping.class);
 
 		 String serverHost = "";
 		 if(getMapping != null) {
 		 	serverHost = StringUtils.join(getMapping.value(), "/");
 		 }
 		 else if(postMapping != null) {
 		 	serverHost = StringUtils.join(postMapping.value(), "/");
 		 }
 		 else if(putMapping != null) {
 		 	serverHost = StringUtils.join(putMapping.value(), "/");
 		 }
 		 else if(deleteMapping != null) {
 		 	serverHost = StringUtils.join(deleteMapping.value(), "/");
 		 }
 		 else if(requestMapping != null) {
 		 	serverHost = StringUtils.join(requestMapping.value(), "/");
 		 }
 		 
 		 Map<String, Object> requestMap = new HashMap<String, Object>();
 		 List<String> requestList = new ArrayList<String>();
 		 
 		 Object[] args = joinPoint.getArgs();
 		 for(Object arg : args) {
 		 	requestList.add(arg.toString());
 		 }
 		 
 		 requestMap.put("params", requestList);
 		 
 		 HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
 		 
 		 traceHolder.setProfile(profile);
 		 traceHolder.setRpcType(RpcTypeEnum.HTTP.name());
 			traceHolder.setServiceCategory("feign");
 			traceHolder.setServiceName(feignClient.name());
 			traceHolder.setMethodName(methodName);
 			traceHolder.setRequestParam(gson.toJson(requestMap));
 			traceHolder.setServiceHost(serverHost);
 			traceHolder.setClientHost(HostUtil.getIP(request));
 			
 			LuminaryRibbonRule.regist(traceHolder);
 			
 			preHandle(traceHolder);
 			request.setAttribute(TraceInfo.TRACE_ID_KEY, traceHolder.getEntity().getTraceId());
 			request.setAttribute(TraceInfo.RPC_ID_KEY, traceHolder.getEntity().getRpcId());
 			result = joinPoint.proceed();
 			traceHolder.getEntity().setResponseInfo(result.toString());
 			postHandle(traceHolder);
 		
 		} catch (Exception e) {
 			log.error(e.getMessage(), e);
 			exceptionHandle(traceHolder, e);
 		}
 			
 		return result;
 		
 	}
 	
 }

[https://github.com/wulinfeng2/luminary-component][https_github.com_wulinfeng2_luminary-component]

[springCloud_OAuth2_JWT_feign_hystrix]: https://blog.csdn.net/guduyishuai/article/details/80865931
[70]: /images/20220518/eceb6efc2a5a4d0abcc49b875a1b5515.png
[70 1]: /images/20220518/f951fb0810e143df9aa3df1ce29c4d0a.png
[https_github.com_wulinfeng2_luminary-component]: https://github.com/wulinfeng2/luminary-component