阿里TTL（Transmittable Thread Local）实现原理

TTL简介TTL全称Transmittable Thread Local是阿里开源的一个用于解决线程池场景下ThreadLocal变量无法在子线程中继承的问题。TTL核心原理简介首先我们知道InheritableThreadLocal允许在创建子线程时子线程继承复制父线程的值但这个继承操作是在构造方法中实现的也就是线程第一次创建的时候实现的。因为这个特点在线程池场景下InheritableThreadLocal子线程就没有办法继承父线程的值了。那么TTL是怎么做到的在介绍原理之前我们先来简单的使用一下InheritableThreadLocal。TTL使用案例在下面的代码中1、首先我们定义了一个TransmittableThreadLocal的context变量2、我们创建了只有一个线程的线程池并提交了一个任务模拟线程池中的线程已经被创建3、后面我们在主线程中往context中设置一个值4、后面提交了俩次任务。第一次提交使用未包装的任务子线程无法获取到主线程设置的值。第二次提交使用TtlRunnable对任务进行了包装子线程成功获取到了值5、在第二次任务中子线程修改了context中的值但最终主线程的值并不会受到影响publicclassTTLDemo{// 1. 定义TTL变量privatestaticfinalTransmittableThreadLocalStringcontextnewTransmittableThreadLocal();publicstaticvoidmain(String[]args)throwsInterruptedException,ExecutionException{ExecutorServiceexecutorExecutors.newFixedThreadPool(1);executor.submit(()-System.out.println(初始化线程Thread.currentThread())).get();// 2. 主线程设置值context.set(Global-Context-Value);System.out.println([Main] 设置值: context.get());// 3. 提交任务未包装- 无法传递值executor.submit(()-System.out.println([Task-RAW] 读取值: context.get())// 输出 null);// 4. 创建可传递任务Runnabletask()-{System.out.println([Task-TTL] 读取值: context.get());// 子线程修改值不会影响父线程context.set(Sub-Thread-Modified);System.out.println([Task-TTL] 修改后: context.get());};// 5. ⭐ 关键包装任务RunnablettlTaskTtlRunnable.get(task);// 6. 提交包装后的任务executor.submit(ttlTask);// 7. 主线程值保持不变Thread.sleep(500);System.out.println([Main] 最终值: context.get());// 仍为 Global-Context-Valueexecutor.shutdown();}}在上面的案例中我们提交任务前需要使用TtlRunnable.get()对要提交的任务包装一层这样ttl才能生效。但如果每次都需要我们自己包装就有点麻烦。这时候我们可以使用自动包装线程池由线程池自动来帮助我们实现任务的包装。修改后的代码如下使用TtlExecutors.getTtlExecutorService()对我们的线程池进行包装。// 1. 定义TTL变量privatestaticfinalTransmittableThreadLocalStringcontextnewTransmittableThreadLocal();publicstaticvoidmain(String[]args)throwsInterruptedException,ExecutionException{ExecutorServiceexecutorTtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(1));executor.submit(()-System.out.println(初始化线程Thread.currentThread())).get();// 2. 主线程设置值context.set(Global-Context-Value);System.out.println([Main] 设置值: context.get());// 3. 提交任务- 能够读取到主线程设置的值executor.submit(()-System.out.println([Task-RAW] 读取值: context.get()));// 4. 主线程值保持不变Thread.sleep(500);System.out.println([Main] 最终值: context.get());// 仍为 Global-Context-Valueexecutor.shutdown();}实现原理也很简单就是我们提交任务的时候它帮助我们使用TtlRunable.get()包装了一下我们提交的任务。ttl原理解析知道了ttl怎么使用后我们就可以开始研究他的实现原理了。在陷入细节之前我们先看下整体流程。整体工作流程示意图--------------------- --------------------- | 任务提交线程 | | 线程池线程 | | (父线程) | | (执行线程) | -------------------- -------------------- | | | 1. capture() | | 创建 captured 快照 | | | | 2. 提交 TtlRunnable | --------------------------- | | 3. replay(captured) | | a. 备份当前状态 → backup | | b. 应用 captured 快照 | | | | 4. 执行 runnable.run() | | (使用父线程的上下文) | | | | 5. restore(backup) | | (恢复原始状态) | | -------------------- --------------------首先在提交任务之前会创建一个快照这个快照包含了父线程此刻所有的TTL数据提交任务后子线程执行时首先会进行replay。replay做俩件事2.1. 备份当前线程TTL数据状态2.2 应用快照。应用快照后就相当于是子线程继承了父线程的上下文了执行任务执行任务后线程恢复原始状态了解了整体流程后我们来看下具体的实现逻辑。首先我们来看一下TtlRunnable.get()方法做了什么publicstaticTtlRunnableget(NullableRunnablerunnable,booleanreleaseTtlValueReferenceAfterRun,booleanidempotent){if(runnablenull)returnnull;if(runnableinstanceofTtlEnhanced){// avoid redundant decoration, and ensure idempotencyif(idempotent)return(TtlRunnable)runnable;elsethrownewIllegalStateException(Already TtlRunnable!);}returnnewTtlRunnable(runnable,releaseTtlValueReferenceAfterRun);}其实里面没有太多逻辑就是new了一个TtlRunnable。所以我们需要看下TtlRunnable构造方法中做了哪些事情privateTtlRunnable(NonNullRunnablerunnable,booleanreleaseTtlValueReferenceAfterRun){this.capturedRefnewAtomicReference(capture());this.runnablerunnable;this.releaseTtlValueReferenceAfterRunreleaseTtlValueReferenceAfterRun;}在TtlRunnable的构造方法中执行了一个capture()方法这个就是我们之前说的捕获快照即获取当前线程所有TTL的值。capture()遍历了transmitteeSet调用transmittee的capture进行快照捕获。publicstaticObjectcapture(){finalHashMapTransmitteeObject,Object,Objecttransmittee2ValuenewHashMap(transmitteeSet.size());for(TransmitteeObject,Objecttransmittee:transmitteeSet){try{transmittee2Value.put(transmittee,transmittee.capture());}catch(Throwablet){if(logger.isLoggable(Level.WARNING)){logger.log(Level.WARNING,exception when Transmitter.capture for transmittee transmittee(class transmittee.getClass().getName()), just ignored; cause: t,t);}}}returnnewSnapshot(transmittee2Value);}transmitteeSets是一个静态变量在static中初始化了俩个transmitteeprivatestaticfinalSetTransmitteeObject,ObjecttransmitteeSetnewCopyOnWriteArraySet();static{registerTransmittee(ttlTransmittee);registerTransmittee(threadLocalTransmittee);}这里我们主要看ttlTransmittee这个是TTL专用的threadLocalTransmittee是兼容ThreadLocal的如果我们项目中已有ThreadLocal想拥有TransmittableThreadLocal 的传递能力又不想替换成TransmittableThreadLocal threadLocalTransmittee就登场了但我们一般用不到所以这里我们不多介绍。publicHashMapTransmittableThreadLocalObject,Objectcapture(){finalHashMapTransmittableThreadLocalObject,Objectttl2ValuenewHashMap(holder.get().size());for(TransmittableThreadLocalObjectthreadLocal:holder.get().keySet()){ttl2Value.put(threadLocal,threadLocal.copyValue());}returnttl2Value;}实现逻辑很简单就是遍历holderholder是一个全局变量保存了所有的TransmittableThreadLocal也就是说这里会遍历所有的TransmittableThreadLocal的copyValue()实现拷贝这里的copyValue()默认其实就是返回的TransmittableThreadLocal.get()的值也就是说默认是浅拷贝如果要实现深拷贝可以重写copy()方法。这里在扩展一下holder是怎么维护所有的TransmittableThreadLocal的其实我们在调用TransmittableThreadLocal.set()方法的时候就会把当前的TransmittableThreadLocal加入到holder中publicfinalvoidset(Tvalue){if(!disableIgnoreNullValueSemanticsvaluenull){// may set null to remove valueremove();}else{super.set(value);addThisToHolder();// 将当前TransmittableThreadLocal加入到holder中}}ok也就是说在TtlRunnable构造方法中会创建一个快照快照保存当前线程下的所有TransmittableThreadLocal值那这个快照什么时候用到呢这就要看TtlRunnable的run方法了。在run方法中1、第一步是获取父线程的快照信息。2、将快照中的信息应用到当前线程relpay简单来说就是将父线程所有TransmittableThreadLocal的值都“拷贝”到当前线程从而实现父子线程的ThreadLocal值传递。Overridepublicvoidrun(){finalObjectcapturedcapturedRef.get();if(capturednull||releaseTtlValueReferenceAfterRun!capturedRef.compareAndSet(captured,null)){thrownewIllegalStateException(TTL value reference is released after run!);}finalObjectbackupreplay(captured);try{runnable.run();}finally{restore(backup);}}我们重点来看看这个replay。首先遍历holder将所有的TTL做一个备份备份的是当前线程的数据在遍历的过程中如果存在快照没有的TTL那么这里会做个移除。通常这种场景存在在线程池中线程在执行其他任务的时候新增了一个TTL如果不移除的化当前任务可能会受到其他任务的影响。setTtlValuesTo。这是一个关键的方法该方法中将任务快照到内容应用到了当前线程。确保任务执行时当前线程所有TTL的值与任务创建时的一致。publicstaticObjectreplay(NonnullObjectcaptured){SuppressWarnings(unchecked)MapTransmittableThreadLocal?,ObjectcapturedMap(MapTransmittableThreadLocal?,Object)captured;MapTransmittableThreadLocal?,ObjectbackupnewHashMapTransmittableThreadLocal?,Object();for(Iterator?extendsMap.EntryTransmittableThreadLocal?,?iteratorholder.get().entrySet().iterator();iterator.hasNext();){Map.EntryTransmittableThreadLocal?,?nextiterator.next();TransmittableThreadLocal?threadLocalnext.getKey();// backupbackup.put(threadLocal,threadLocal.get());// clear the TTL values that is not in captured// avoid the extra TTL values after replay when run task// 移除快照中没有的TTLif(!capturedMap.containsKey(threadLocal)){iterator.remove();threadLocal.superRemove();}}// set values to captured TTLsetTtlValuesTo(capturedMap);// call beforeExecute callbackdoExecuteCallback(true);returnbackup;}privatestaticvoidsetTtlValuesTo(NonnullMapTransmittableThreadLocal?,ObjectttlValues){for(Map.EntryTransmittableThreadLocal?,Objectentry:ttlValues.entrySet()){SuppressWarnings(unchecked)TransmittableThreadLocalObjectthreadLocal(TransmittableThreadLocalObject)entry.getKey();threadLocal.set(entry.getValue());}}最后看一看任务执行完后的restore方法。restore的逻辑就比较简单了就是通过备份将状态恢复成执行前的状态。重点还是在setTtlValuesTo中。publicstaticvoidrestore(NonnullObjectbackup){SuppressWarnings(unchecked)MapTransmittableThreadLocal?,ObjectbackupMap(MapTransmittableThreadLocal?,Object)backup;// call afterExecute callbackdoExecuteCallback(false);for(Iterator?extendsMap.EntryTransmittableThreadLocal?,?iteratorholder.get().entrySet().iterator();iterator.hasNext();){Map.EntryTransmittableThreadLocal?,?nextiterator.next();TransmittableThreadLocal?threadLocalnext.getKey();// clear the TTL values that is not in backup// avoid the extra TTL values after restoreif(!backupMap.containsKey(threadLocal)){iterator.remove();threadLocal.superRemove();}}// restore TTL valuessetTtlValuesTo(backupMap);}privatestaticvoidsetTtlValuesTo(NonnullMapTransmittableThreadLocal?,ObjectttlValues){for(Map.EntryTransmittableThreadLocal?,Objectentry:ttlValues.entrySet()){SuppressWarnings(unchecked)TransmittableThreadLocalObjectthreadLocal(TransmittableThreadLocalObject)entry.getKey();threadLocal.set(entry.getValue());}}ok到这里我们就将源码看完了可以在回头看一下流程图加深一下理解。常见的坑没有使用TtlExecutors.getTtlExecutorService对线程池进行包装也没有用TtlRunnable.get()这听上去比较初级的错误但是在美团的某个团队中重复踩了很多次这个坑。因为封装Context的可能不是你然后你看其他地方都这么用你也这么用。但是你忽略了一个问题其他人的场景下可能没用到线程池所以他们不会有问题但是你用到了线程池所以你出问题了。总之一旦用到了上下文一定要格外小心线程池的场景。浅拷贝默认情况下TransmittableThreadLocal是浅拷贝如果是个对象的话子线程修改了值会影响到父线程。我们可以重写TransmittableThreadLocal的copy方法。如下// 1. 定义TTL变量privatestaticfinalTransmittableThreadLocalUsercontextnewTransmittableThreadLocal(){OverridepublicUsercopy(UserparentValue){returnnewUser(parentValue.getName());}};publicstaticvoidmain(String[]args)throwsInterruptedException,ExecutionException{ExecutorServiceexecutorTtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(1));executor.submit(()-System.out.println(初始化线程Thread.currentThread())).get();// 2. 主线程设置值UserusernewUser(小明);context.set(user);System.out.println([Main] 设置值: context.get());// 3. 提交任务- 能够读取到主线程设置的值executor.submit(()-{context.get().setName(小王);System.out.println([Task-RAW] 读取值: context.get());});// 4. 因为深拷贝了。所以主线程值保持不变Thread.sleep(500);System.out.println([Main] 最终值: context.get());// 仍为 Global-Context-Valueexecutor.shutdown();}但在深拷贝之前我们可以先回答自己一个问题我们真的要允许子线程去修改我们的值吗如果不允许的话可以将对象设计成不可变的。AllArgsConstructorDatapublicclassUser{privateStringname;}