帮彩票网站做流量提升,建设阿里巴巴网站,利用js做网站,phpcms 多语言网站目录
1、前言
2、TransmittableThreadLocal
2.1、使用场景
2.2、基本使用
3、实现原理
4、小结 1、前言
书接上回《【JUC进阶】13. InheritableThreadLocal》#xff0c;提到了InheritableThreadLocal虽然能进行父子线程的值传递#xff0c;但是如果在线程池中#x…目录
1、前言
2、TransmittableThreadLocal
2.1、使用场景
2.2、基本使用
3、实现原理
4、小结 1、前言
书接上回《【JUC进阶】13. InheritableThreadLocal》提到了InheritableThreadLocal虽然能进行父子线程的值传递但是如果在线程池中就无法达到预期的效果了。为了更好的解决该问题TransmittableThreadLocal诞生了。
2、TransmittableThreadLocal
TransmittableThreadLocal 是Alibaba开源的、用于解决 “在使用线程池等会缓存线程的组件情况下传递ThreadLocal” 问题的 InheritableThreadLocal 扩展。既然是扩展那么自然具备InheritableThreadLocal不同线程间值传递的能力。但是他也是专门为了解决InheritableThreadLocal在线程池中出现的问题的。
官网地址https://github.com/alibaba/transmittable-thread-local
2.1、使用场景
分布式跟踪系统 或 全链路压测即链路打标日志收集记录系统上下文Session级Cache应用容器或上层框架跨应用代码给下层SDK传递信息
2.2、基本使用
我们拿《【JUC进阶】13. InheritableThreadLocal》文中最后的demo进行改造。这里需要配合TtlExecutors一起使用。这里先讲述使用方法具体为什么下面细说。
首先我们需要添加依赖
!-- https://mvnrepository.com/artifact/com.alibaba/transmittable-thread-local --
dependencygroupIdcom.alibaba/groupIdartifactIdtransmittable-thread-local/artifactIdversion2.14.2/version
/dependency
其次ThreadLocal的实现改为TransmittableThreadLocal。
static ThreadLocalString threadLocal new TransmittableThreadLocal();
最后创建线程池的时候使用TTL装饰器
static ExecutorService executorService TtlExecutors.getTtlExecutorService(Executors.newSingleThreadExecutor());
完整代码如下
// threadlocal改为TransmittableThreadLocal
static ThreadLocalString threadLocal new TransmittableThreadLocal();// 线程池添加TtlExecutors
static ExecutorService executorService TtlExecutors.getTtlExecutorService(Executors.newSingleThreadExecutor());public static void main(String[] args) throws InterruptedException {//threadLocal.set(我是主线程的threadlocal变量变量值为000000);// 线程池执行子线程executorService.submit(() - {System.out.println(----- 子线程 Thread.currentThread() ----- 获取threadlocal变量 threadLocal.get());});// 主线程睡眠3s模拟运行Thread.sleep(3000);// 将变量修改为11111在InheritableThreadLocal中修改是无效的threadLocal.set(我是主线程的threadlocal变量变量值为11111);// 这里线程池重新执行线程任务executorService.submit(() - {System.out.println(----- 子线程 Thread.currentThread() ----- 获取threadlocal变量 threadLocal.get());});// 线程池关闭executorService.shutdown();
}
执行看下效果 已经成功获取到threadlocal变量。 该方式也解决了因为线程被重复利用而threadlocal重新赋值失效的问题。 3、实现原理
首先可以看到TransmittableThreadLocal继承InheritableThreadLocal同时实现了TtlCopier接口。TtlCopier接口只提供了一个方法copy()。看到这里可能有人大概猜出来他的实现原理了既然实现了copy()方法那么大概率是将父线程的变量复制一份存起来接着找个地方存起来然后找个适当的时机再还回去。没错其实就是这样。
public class TransmittableThreadLocalT extends InheritableThreadLocalT implements TtlCopierT {
}
知道了TransmittableThreadLocal类的定义之后我们再来看一个重要的属性holder
// Note about the holder:
// 1. holder self is a InheritableThreadLocal(a *ThreadLocal*).
// 2. The type of value in the holder is WeakHashMapTransmittableThreadLocalObject, ?.
// 2.1 but the WeakHashMap is used as a *Set*:
// the value of WeakHashMap is *always* null, and never used.
// 2.2 WeakHashMap support *null* value.
private static final InheritableThreadLocalWeakHashMapTransmittableThreadLocalObject, ? holder new InheritableThreadLocalWeakHashMapTransmittableThreadLocalObject, ?() {Overrideprotected WeakHashMapTransmittableThreadLocalObject, ? initialValue() {return new WeakHashMap();}Overrideprotected WeakHashMapTransmittableThreadLocalObject, ? childValue(WeakHashMapTransmittableThreadLocalObject, ? parentValue) {return new WeakHashMap(parentValue);}};
这里存放的是一个全局的WeakMap同ThreadLocal一样weakMap也是为了解决内存泄漏的问题里面存放了TransmittableThreadLocal对象并且重写了initialValue和childValue方法尤其是childValue可以看到在即将异步时父线程的属性是直接作为初始化值赋值给子线程的本地变量对象。引入holder变量后也就不必对外暴露Thread中的 inheritableThreadLocals保持ThreadLocal.ThreadLocalMap的封装性。
而TransmittableThreadLocal中的get()和set()方法都是从该holder中获取或添加该map。
重点来了前面不是提到了需要借助于TtlExecutors.getTtlExecutorService()包装线程池才能达到效果吗我们来看看这里做了什么事。
我们从TtlExecutors.getTtlExecutorService()方法跟进可以发现一个线程池的ttl包装类ExecutorServiceTtlWrapper。其中包含了我们执行线程的方法submit()和execute()。我们进入submit()方法
NonNull
Override
public T FutureT submit(NonNull CallableT task) {return executorService.submit(TtlCallable.get(task, false, idempotent));
}
可以发现在线程池进行任务执行时对我们提交的任务进行了一层预处理TtlCallable.get()。TtlCallable也是Callable的装饰类同样还有TtlRunnable也是同样道理。我们跟进该方法偷瞄一眼
Nullable
Contract(value null, _, _ - null; !null, _, _ - !null, pure true)
public static T TtlCallableT get(Nullable CallableT callable, boolean releaseTtlValueReferenceAfterCall, boolean idempotent) {if (callable null) return null;if (callable instanceof TtlEnhanced) {// avoid redundant decoration, and ensure idempotencyif (idempotent) return (TtlCallableT) callable;else throw new IllegalStateException(Already TtlCallable!);}return new TtlCallable(callable, releaseTtlValueReferenceAfterCall);
}
上面判断下当前线程的类型是否已经是TtlEnhanced如果是直接返回否则创建一个TtlCallable。接着进入new TtlCallable()方法
private TtlCallable(NonNull CallableV callable, boolean releaseTtlValueReferenceAfterCall) {this.capturedRef new AtomicReference(capture());this.callable callable;this.releaseTtlValueReferenceAfterCall releaseTtlValueReferenceAfterCall;
}
可以看到在初始化线程的时候调用了一个capture()方法并将该方法得到的值存放在capturedRef中。没错这里就是上面我们提到的将父线程的本地变量复制一份快照存放起来。跟进capture()
NonNull
public static Object capture() {final HashMapTransmitteeObject, Object, Object transmittee2Value newHashMap(transmitteeSet.size());for (TransmitteeObject, Object transmittee : transmitteeSet) {try {transmittee2Value.put(transmittee, transmittee.capture());} catch (Throwable t) {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);}}}return new Snapshot(transmittee2Value);
}
这里的transmitteeSet是一个存放Transmitteedede 集合在初始化中会将我们 前面提到的holder注册进去
private static final SetTransmitteeObject, Object transmitteeSet new CopyOnWriteArraySet();static {registerTransmittee(ttlTransmittee);registerTransmittee(threadLocalTransmittee);
}SuppressWarnings(unchecked)
public static C, B boolean registerTransmittee(NonNull TransmitteeC, B transmittee) {return transmitteeSet.add((TransmitteeObject, Object) transmittee);
}
跟进transmittee.capture()方法该方法由静态内部类Transmitter实现并重写com.alibaba.ttl.TransmittableThreadLocal.Transmitter.Transmittee#capture
private static final TransmitteeHashMapTransmittableThreadLocalObject, Object, HashMapTransmittableThreadLocalObject, Object ttlTransmittee new TransmitteeHashMapTransmittableThreadLocalObject, Object, HashMapTransmittableThreadLocalObject, Object() {NonNullOverridepublic HashMapTransmittableThreadLocalObject, Object capture() {final HashMapTransmittableThreadLocalObject, Object ttl2Value newHashMap(holder.get().size());for (TransmittableThreadLocalObject threadLocal : holder.get().keySet()) {ttl2Value.put(threadLocal, threadLocal.copyValue());}return ttl2Value;}
}
transmittee.capture()扫描holder里目前存放的k-v里的key就是需要传给子线程的TTL对象其中调用的threadLocal.copyValue()便是前面看到的TtlCopier接口提供的方法。
看到这里已经大致符合我们前面的猜想将变量复制一份存起来。那么不出意外接下来应该就是要找个适当的机会还回去。我们接着看。
接下来我们看真正执行线程的时候也就是call()方法。由于前面线程被TtlCallable包装过以为这里的call()方法肯定是TtlCallable.call()
Override
SuppressFBWarnings(THROWS_METHOD_THROWS_CLAUSE_BASIC_EXCEPTION)
public V call() throws Exception {// 获取由之前捕获到的父线程变量集final Object captured capturedRef.get();if (captured null || releaseTtlValueReferenceAfterCall !capturedRef.compareAndSet(captured, null)) {throw new IllegalStateException(TTL value reference is released after call!);}// 这里的backup是当前线程原有的变量这里进行备份等线程执行完毕后会将该变量进行恢复final Object backup replay(captured);try {// 任务执行return callable.call();} finally {// 恢复上述提到的backup原有变量restore(backup);}
}
果然在执行线程时先获取之前存放起来的变量。然后调用replay()
NonNull
public static Object replay(NonNull Object captured) {final Snapshot capturedSnapshot (Snapshot) captured;final HashMapTransmitteeObject, Object, Object transmittee2Value newHashMap(capturedSnapshot.transmittee2Value.size());for (Map.EntryTransmitteeObject, Object, Object entry : capturedSnapshot.transmittee2Value.entrySet()) {TransmitteeObject, Object transmittee entry.getKey();try {Object transmitteeCaptured entry.getValue();transmittee2Value.put(transmittee, transmittee.replay(transmitteeCaptured));} catch (Throwable t) {if (logger.isLoggable(Level.WARNING)) {logger.log(Level.WARNING, exception when Transmitter.replay for transmittee transmittee (class transmittee.getClass().getName() ), just ignored; cause: t, t);}}}return new Snapshot(transmittee2Value);
}
继续跟进transmittee.replay(transmitteeCaptured)
NonNull
Override
public HashMapTransmittableThreadLocalObject, Object replay(NonNull HashMapTransmittableThreadLocalObject, Object captured) {final HashMapTransmittableThreadLocalObject, Object backup newHashMap(holder.get().size());for (final IteratorTransmittableThreadLocalObject iterator holder.get().keySet().iterator(); iterator.hasNext(); ) {TransmittableThreadLocalObject threadLocal iterator.next();// 这里便是所有原生的本地变量都暂时存储在backup里用于之后恢复用backup.put(threadLocal, threadLocal.get());// clear the TTL values that is not in captured// avoid the extra TTL values after replay when run task// 这里检查如果当前变量不存在于捕获到的线程变量那么就将他清除掉对应线程的本地变量也清理掉// 为什么要清除因为从使用这个子线程做异步那里捕获到的本地变量并不包含原生的变量当前线程// 在做任务时的首要目标是将父线程里的变量完全传递给任务如果不清除这个子线程原生的本地变量// 意味着很可能会影响到任务里取值的准确性。这也就是为什么上面需要做备份的原因。if (!captured.containsKey(threadLocal)) {iterator.remove();threadLocal.superRemove();}}// set TTL values to capturedsetTtlValuesTo(captured);// call beforeExecute callbackdoExecuteCallback(true);return backup;
}
继续跟进setTtlValuesTo(captured)这里就是把父线程本地变量赋值给当前线程了
private static void setTtlValuesTo(NonNull HashMapTransmittableThreadLocalObject, Object ttlValues) {for (Map.EntryTransmittableThreadLocalObject, Object entry : ttlValues.entrySet()) {TransmittableThreadLocalObject threadLocal entry.getKey();threadLocal.set(entry.getValue());}
}
到这里基本的实现原理也差不多了基本和我们前面猜想的一致。但是这里还少了前面提到的backup变量如何恢复的步骤既然到这里了一起看一下跟进restore(backup)
public static void restore(NonNull Object backup) {for (Map.EntryTransmitteeObject, Object, Object entry : ((Snapshot) backup).transmittee2Value.entrySet()) {TransmitteeObject, Object transmittee entry.getKey();try {Object transmitteeBackup entry.getValue();transmittee.restore(transmitteeBackup);} catch (Throwable t) {if (logger.isLoggable(Level.WARNING)) {logger.log(Level.WARNING, exception when Transmitter.restore for transmittee transmittee (class transmittee.getClass().getName() ), just ignored; cause: t, t);}}}
}
继续看transmittee.restore(transmitteeBackup)
Override
public void restore(NonNull HashMapTransmittableThreadLocalObject, Object backup) {// call afterExecute callbackdoExecuteCallback(false);for (final IteratorTransmittableThreadLocalObject iterator holder.get().keySet().iterator(); iterator.hasNext(); ) {TransmittableThreadLocalObject threadLocal iterator.next();// clear the TTL values that is not in backup// avoid the extra TTL values after restoreif (!backup.containsKey(threadLocal)) {iterator.remove();threadLocal.superRemove();}}// restore TTL valuessetTtlValuesTo(backup);
}
与replay类似只是重复进行了将backup赋给当前线程的步骤。到此基本结束。附上官网的时序图帮助理解 4、小结
所以总结下来TransmittableThreadLocal的实现原理主要就是依赖于TtlRunnable或TtlCallable装饰类的预处理方法TtlExecutors是将普通线程转换成Ttl包装的线程而ttl包装的线程会进行本地变量的预处理也就是capture()拷贝一份快照到内存中然后通过replay方法将父线程的变量赋值给当前线程。
