ZipKin原理学习--Zipkin多线程及线程池中追踪一致性问题解决

喜欢ヅ旅行 2022-05-24 03:39 258阅读 0赞

在学习Zipkin分布式追踪系统中我们了解到Trace在整个调用链是一致的，在web服务中可以通过在header设置Trace值在不同的服务中进行传递，那样在一个服务内部不同的线程，甚至是线程池中Zipkin是如何处理的，接下来我们来了解学习一下。

##### 单个线程 #####

在单个线程的调用过程中，我们一般都知道通过ThreadLocal来完成在整个线程执行过程中获取相同的Trace值，Zipkin也是通过定义了一个ThreadLocal<TraceContext> local来实现处理的。

##### 父子线程 #####

在主线程中新建立一个子线程时使用ThreadLocal就无效了，因此Zipkin提供了如下定义方式，使用InheritableThreadLocal定义（可以参考博客[Java 多线程：InheritableThreadLocal 实现原理][Java _InheritableThreadLocal]）

static final InheritableThreadLocal<TraceContext> INHERITABLE = new InheritableThreadLocal<>();

这样就是存在父子线程，在创建子线程的过程中会将父线程的值全部拷贝到子线程中，这样在子线程中依然可以获取到Trace值，因此如下面的代码追踪链路依然是完整的。

@RequestMapping("/start2")
        public String start(HttpServletRequest request1,HttpServletResponse response1) throws InterruptedException, IOException {
        	
        	Thread thread = new Thread((new Runnable() {
    			
    			@Override
    			public void run() {
    				System.err.println(Thread.currentThread().hashCode());
    				data = restTemplate.getForObject("http://localhost:9090/foo", String.class);
    			}
    		}));
        	thread.start();
        	
        	return data;
        }

##### 线程池 #####

在我们新创建一个线程，然后将线程提交给线程池时，由于线程池中线程执行的原理此时原线程中的ThreadLocal和InheritableThreadLocal都是无效的，追踪Trace值因此会丢失，导致整个调用链出现断路，如下面代码。

@RequestMapping("/start2")
        public String start(HttpServletRequest request1,HttpServletResponse response1) throws InterruptedException, IOException {
    		String data = "";
        	Thread thread = new Thread((new Runnable() {
    			
    			@Override
    			public void run() {
    				System.err.println(Thread.currentThread().hashCode());
    				data = restTemplate.getForObject("http://localhost:9090/foo", String.class);
    			}
    		}));
        	executor.execute(thread);
        	Thread.sleep(10000);
        	
        	return data;
        }

目前Zipkin类CurrentTraceContext给出对线程及线程池的的处理方法就是实现了Runnable重新实现了run方法，这样就解决了线程池的问题，当然不只提供了创建线程的方法，还包括线程池和Callable

public Runnable wrap(Runnable task) {
    	//获取父线程中的Trace
        final TraceContext invocationContext = get();
        class CurrentTraceContextRunnable implements Runnable {
          @Override public void run() {
    		//将父线程中的Trace复制到子线程中
            try (Scope scope = maybeScope(invocationContext)) {
              task.run();
            }
          }
        }
        return new CurrentTraceContextRunnable();
      }
      
      public Scope maybeScope(@Nullable TraceContext currentSpan) {
        TraceContext currentScope = get();
        if (currentSpan == null) {
          if (currentScope == null) return Scope.NOOP;
          return newScope(null);
        }
        return currentSpan.equals(currentScope) ? Scope.NOOP : newScope(currentSpan);
      }
      
      public Executor executor(Executor delegate) {
        class CurrentTraceContextExecutor implements Executor {
          @Override public void execute(Runnable task) {
            delegate.execute(CurrentTraceContext.this.wrap(task));
          }
        }
        return new CurrentTraceContextExecutor();
      }
    
      /**
       * Decorates the input such that the {@link #get() current trace context} at the time a task is
       * scheduled is made current when the task is executed.
       */
      public ExecutorService executorService(ExecutorService delegate) {
        class CurrentTraceContextExecutorService extends brave.internal.WrappingExecutorService {
    
          @Override protected ExecutorService delegate() {
            return delegate;
          }
    
          @Override protected <C> Callable<C> wrap(Callable<C> task) {
            return CurrentTraceContext.this.wrap(task);
          }
    
          @Override protected Runnable wrap(Runnable task) {
            return CurrentTraceContext.this.wrap(task);
          }
        }
        return new CurrentTraceContextExecutorService();
      }
      public <C> Callable<C> wrap(Callable<C> task) {
        final TraceContext invocationContext = get();
        class CurrentTraceContextCallable implements Callable<C> {
          @Override public C call() throws Exception {
            try (Scope scope = maybeScope(invocationContext)) {
              return task.call();
            }
          }
        }
        return new CurrentTraceContextCallable();
      }

[Java _InheritableThreadLocal]: https://blog.csdn.net/qq924862077/article/details/80279330