关于threadlocal:阿里-TransmittableThreadLocal-代码简读

零 筹备

0 FBI WARNING

文章异样啰嗦且绕弯。

1 TransmittableThreadLocal 是什么

当开发人员须要在线程池的线程中传递某些参数的时候，jdk 的 ThreadLocal 很难实现，动态变量则会面临不够灵便和呈现线程平安等问题。
TransmittableThreadLocal 是阿里开源工具包，用于解决这一问题。

2 版本

• jdk 版本

Azul JDK 17.0.2

• transmittable-thread-local

2.13.0-Beta1

• junit-jupiter

5.8.2

一 Demo

import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.threadpool.TtlExecutors;
import org.junit.jupiter.api.Test;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TreadLocalTest {

    @Test
    public void transmittableThreadLocal() {TransmittableThreadLocal<Integer> tl = new TransmittableThreadLocal<>();

        tl.set(6);
        System.out.println("父线程获取数据：" + tl.get()); // 第一次输入：6

        // 应用 jdk 的 Executors 工具创立一个线程池
        // 留神，这个线程池里只有一个线程
        Executor realPool = Executors.newFixedThreadPool(1);
        
        // 应用 TtlExecutors 创立一个 Ttl 框架封装的线程池
        Executor pool = TtlExecutors.getTtlExecutor(realPool);

        // 应用线程池跑一个工作
        pool.execute(() -> {Integer i = tl.get();
            System.out.println("第一次获取数据：" + i); // 第二次输入：6
        });

        // 批改一下 tl 里的值，并再跑一次工作
        tl.set(7);
        pool.execute(() -> {Integer i = tl.get();
            System.out.println("第二次获取数据：" + i); // 第三次输入：7
        });
    }
}

二 先从 InheritableThreadLocal 说起

1 Thread

InheritableThreadLocal 是 jdk 中自带的 ThreadLocal 的子类，在 jdk 的 Thread 对象中，会对它有独自的反对。
首先来看 Thread 的构造方法：

// java.lang.Thread 的外围构造方法
private Thread(ThreadGroup g, Runnable target, String name,
                   long stackSize, AccessControlContext acc,
                   boolean inheritThreadLocals) {
    
    // 此处省略一大段无关代码...
    
    // inheritThreadLocals 是一个 boolean 类型的值，是一个“是否启用 inheritableThreadLocals”的开关
    // parent 是发明此线程的父线程
    if (inheritThreadLocals && parent.inheritableThreadLocals != null)
        // 如果父线程的 inheritableThreadLocals 存在，则此处会将它挪到以后线程里
        // ThreadLocal.createInheritedMap 是一个深拷贝，会创立新的 Entry
        this.inheritableThreadLocals = ThreadLocal.createInheritedMap(parent.inheritableThreadLocals);
    
    // 此处省略一大段无关代码...
}

2 ThreadLocalMap

来看一下 ThreadLocal.createInheritedMap：

// java.lang.ThreadLocal
static ThreadLocalMap createInheritedMap(ThreadLocalMap parentMap) {return new ThreadLocalMap(parentMap);
}

这个办法会创立一个 ThreadLocalMap，再来追踪一下 ThreadLocalMap 的结构器：
（值得注意的是，ThreadLocalMap 是 ThreadLocal 的外部类，所以其实代码逻辑还是在 ThreadLocal.java 中）

// java.lang.ThreadLocal
private ThreadLocalMap(ThreadLocalMap parentMap) {Entry[] parentTable = parentMap.table;
    int len = parentTable.length;
    setThreshold(len);
    table = new Entry[len];

    // 此处把 ThreadLocalMap 里的元素都遍厉一遍
    // 而后都创立成新的 Entry 并塞到新的 ThreadLocalMap 里
    for (Entry e : parentTable) {if (e != null) {
            // 此处获取了 Entry 的 key，实质上就是 ThreadLocal 自身
            ThreadLocal<Object> key = (ThreadLocal<Object>) e.get();
            if (key != null) {
                // 用 key 取获取 value，这行代码重点关注，下文会提到
                Object value = key.childValue(e.value);
                // 此处创立新的 Entry
                Entry c = new Entry(key, value);
                // 解决 hash 碰撞问题并存入
                int h = key.threadLocalHashCode & (len - 1);
                while (table[h] != null)
                    h = nextIndex(h, len);
                table[h] = c;
                size++;
            }
        }
    }
}

3 childValue

这里须要重点关注一行代码：

Object value = key.childValue(e.value);

这个办法是 ThreadLocal 中的：

// java.lang.ThreadLocal
T childValue(T parentValue) {throw new UnsupportedOperationException();
}

由上文可见，这是一个没有被实现的预留模板办法。在 InheritableThreadLocal 中对其进行了实现：

// java.lang.InheritableThreadLocal
protected T childValue(T parentValue) {return parentValue;}

5 initialValue

initialValue 同样是 ThreadLocal 提供的一个空办法：

// java.lang.ThreadLocal
protected T initialValue() {return null;}

这个办法会作用在 ThreadLocal 的 get() 办法里：

// step 1
// java.lang.ThreadLocal
public T get() {Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null) {ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null) {
            // 如果 Entry 存在，则此处会返回 Entry 的 value
            T result = (T)e.value;
            return result;
        }
    }
    // 如果 Entry 不存在，或者 ThreadLocalMap 不存在，会在这里初始化一个 value
    // 这个办法见 step 2
    return setInitialValue();}

// step 2
// java.lang.ThreadLocal
private T setInitialValue() {
    // 这里初始化一个值
    T value = initialValue();
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null) {
        // 将初始化进去的值存进去
        map.set(this, value);
    } else {
        // 初始化 ThreadLocalMap
        createMap(t, value);
    }
    
    // 此处疏忽这段代码
    if (this instanceof TerminatingThreadLocal) {TerminatingThreadLocal.register((TerminatingThreadLocal<?>) this);
    }
    
    // 返回
    return value;
}

5 InheritableThreadLocal 的作用和问题

假如 Thread A 是 Thread B 的父线程，由上述代码可知：

• A 的 InheritableThreadLocal 内的数据能够被 B 继承
• 继承形式是在创立 B 的时候，在构造方法里间接 copy 一份 InheritableThreadLocal 内的元素
• copy 是一个快照机制，一旦完结，再去批改 A 中的 InheritableThreadLocal 中的元素，就不会同步给 B 了

那么问题来了：
如果零碎中须要做到 A 和 B 的 InheritableThreadLocal 实时同步，应该如何解决？

三 TransmittableThreadLocal

先来看下列三行代码：

// 创立一个 TransmittableThreadLocal
ThreadLocal<Integer> tl = new TransmittableThreadLocal<>();

tl.set(6);

Integer i = tl.get();

1 结构器

TransmittableThreadLocal 的结构器非常简单。

// 是否要疏忽 null value，如果这个参数为 false，则哪怕 value 是 null，也会存储下来
private final boolean disableIgnoreNullValueSemantics;

// 这个参数默认为 false
public TransmittableThreadLocal() {this(false);
}


public TransmittableThreadLocal(boolean disableIgnoreNullValueSemantics) {this.disableIgnoreNullValueSemantics = disableIgnoreNullValueSemantics;}

2 holder

holder 是 TransmittableThreadLocal 的动态成员变量，是一个 InheritableThreadLocal。

// com.alibaba.ttl.TransmittableThreadLocal
private static final InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>> holder =
    new InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>>() {// 复写这个办法应该没有别的深意，只是为了避免在调用 holder.get().xxx() 的时候报空指针
    // 应该是开发人员感觉这样比拟优雅
    @Override
    protected WeakHashMap<TransmittableThreadLocal<Object>, ?> initialValue() {return new WeakHashMap<TransmittableThreadLocal<Object>, Object>();
    }

    // 这个办法实现了子线程和父线程之间的信息传递
    @Override
    protected WeakHashMap<TransmittableThreadLocal<Object>, ?> childValue(WeakHashMap<TransmittableThreadLocal<Object>, ?> parentValue) {return new WeakHashMap<TransmittableThreadLocal<Object>, Object>(parentValue);
    }
};

由上述可知：

• holder 是一个记录的 value 是 WeakHashMap<TransmittableThreadLocal> 的 InheritableThreadLocal
• WeakHashMap 的 value 并没有被应用到，能够将其视为一个 WeakHashSet
• holder 复写了 initialValue 和 childValue 两个办法

holder 最重要的办法是 addThisToHolder：

// com.alibaba.ttl.TransmittableThreadLocal
// 如果以后 TransmittableThreadLocal 没有被记录在 holder 中，则会在此处 put 进去
private void addThisToHolder() {if (!holder.get().containsKey(this)) {holder.get().put((TransmittableThreadLocal<Object>) this, null); // WeakHashMap supports null value.
    }
}

同样还有移除办法：

// com.alibaba.ttl.TransmittableThreadLocal
private void removeThisFromHolder() {holder.get().remove(this);
}

3 set

存入 value 的办法。

// com.alibaba.ttl.TransmittableThreadLocal
@Override
public final void set(T value) {if (!disableIgnoreNullValueSemantics && null == value) {
        // 如果 value 是 null，且不疏忽 null value，则此处进入删除逻辑
        remove();} else {
        // 存储逻辑
        super.set(value);
        // 将以后的 TransmittableThreadLocal 注册到 holder 里
        addThisToHolder();}
}

4 get

获取 value 的办法。

// com.alibaba.ttl.TransmittableThreadLocal
@Override
public final T get() {T value = super.get();
    // 尝试注册到 holder
    if (disableIgnoreNullValueSemantics || null != value) 
        addThisToHolder();
    return value;
}

5 Snapshot

Snapshot 是 TransmittableThreadLocal 的外部类，用来寄存以后线程内的 ThreadLocal 和 TransmittableThreadLocal 数据。

// com.alibaba.ttl.TransmittableThreadLocal
private static class Snapshot {
    final HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value;
    final HashMap<ThreadLocal<Object>, Object> threadLocal2Value;

    private Snapshot(HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value, HashMap<ThreadLocal<Object>, Object> threadLocal2Value) {
        this.ttl2Value = ttl2Value;
        this.threadLocal2Value = threadLocal2Value;
    }
}

6 Transmitter

Transmitter 是 TransmittableThreadLocal 的外部类，实质上是一组动态工具。

6.1 获取一个快照

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
public static Object capture() {// captureTtlValues()  会将以后线程的 TransmittableThreadLocal 数据做成一个 HashMap
    // captureThreadLocalValues() 会将以后线程的 ThreadLocal 数据做成一个 HashMap
    return new Snapshot(captureTtlValues(), captureThreadLocalValues());
}

6.1.1 获取 holder 中所有的 TransmittableThreadLocal 数据

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
private static HashMap<TransmittableThreadLocal<Object>, Object> captureTtlValues() {HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = new HashMap<TransmittableThreadLocal<Object>, Object>();
    
    for (TransmittableThreadLocal<Object> threadLocal : holder.get().keySet()) {ttl2Value.put(threadLocal, threadLocal.copyValue());
    }
    
    return ttl2Value;
}

6.1.2 获取 threadLocalHolder 中所有 ThreadLocal 数据

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
private static HashMap<ThreadLocal<Object>, Object> captureThreadLocalValues() {final HashMap<ThreadLocal<Object>, Object> threadLocal2Value = new HashMap<ThreadLocal<Object>, Object>();
    
    for (Map.Entry<ThreadLocal<Object>, TtlCopier<Object>> entry : threadLocalHolder.entrySet()) {final ThreadLocal<Object> threadLocal = entry.getKey();
        final TtlCopier<Object> copier = entry.getValue();

        threadLocal2Value.put(threadLocal, copier.copy(threadLocal.get()));
    }
    
    return threadLocal2Value;
}

6.2 重放

6.2.1 replay

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
// 实质上是对一个 snapshot 进行拷贝
public static Object replay(Object captured) {final Snapshot capturedSnapshot = (Snapshot) captured;
    return new Snapshot(replayTtlValues(capturedSnapshot.ttl2Value), replayThreadLocalValues(capturedSnapshot.threadLocal2Value));
}

6.2.2 replayTtlValues

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
// 实质上是对一个 map 进行深拷贝
private static HashMap<TransmittableThreadLocal<Object>, Object> replayTtlValues(HashMap<TransmittableThreadLocal<Object>, Object> captured) {
    
    // 创立一个新的 map
    HashMap<TransmittableThreadLocal<Object>, Object> backup = new HashMap<TransmittableThreadLocal<Object>, Object>();

    for (final Iterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext();) {TransmittableThreadLocal<Object> threadLocal = iterator.next();

        // 将原来的 map 复制到新的 map 中
        backup.put(threadLocal, threadLocal.get());

        // 此处比拟 holder 和 captured 的 key
        // 如果对应不统一，则将 holder 里的数据清空
        if (!captured.containsKey(threadLocal)) {iterator.remove();
            threadLocal.superRemove();}
    }

    // 将 value 和 key 对应起来
    // 这是一个保底纠错逻辑
    setTtlValuesTo(captured);

    // 这是一个临时没有用的扩大办法
    doExecuteCallback(true);

    return backup;
}

6.2.3 replayThreadLocalValues

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
// 实质上是对一个 map 进行深拷贝
private static HashMap<ThreadLocal<Object>, Object> replayThreadLocalValues(HashMap<ThreadLocal<Object>, Object> captured) {final HashMap<ThreadLocal<Object>, Object> backup = new HashMap<ThreadLocal<Object>, Object>();

    for (Map.Entry<ThreadLocal<Object>, Object> entry : captured.entrySet()) {final ThreadLocal<Object> threadLocal = entry.getKey();
        backup.put(threadLocal, threadLocal.get());

        // threadLocalClearMark 是一个空对象，用于占位
        // 如果此处的 value 就是这个空对象，则此处代表这个 ttl 里的 value 曾经被 clear 了
        final Object value = entry.getValue();
        if (value == threadLocalClearMark) 
            threadLocal.remove();
        else 
            threadLocal.set(value);
    }

    return backup;
}

6.3 复原

6.3.1 restore

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
// 用快照来复原以后线程的 ttl 数据
public static void restore(Object backup) {final Snapshot backupSnapshot = (Snapshot) backup;
    restoreTtlValues(backupSnapshot.ttl2Value);
    restoreThreadLocalValues(backupSnapshot.threadLocal2Value);
}

6.3.2 restoreTtlValues

这个办法与 replayTtlValues(…) 办法比拟像

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
private static void restoreTtlValues(HashMap<TransmittableThreadLocal<Object>, Object> backup) {doExecuteCallback(false);

    for (final Iterator<TransmittableThreadLocal<Object>> iterator = holder.get().keySet().iterator(); iterator.hasNext();) {TransmittableThreadLocal<Object> threadLocal = iterator.next();

        if (!backup.containsKey(threadLocal)) {iterator.remove();
            threadLocal.superRemove();}
    }

    setTtlValuesTo(backup);
}

6.3.3 restoreThreadLocalValues

// com.alibaba.ttl.TransmittableThreadLocal.Transmitter
private static void restoreThreadLocalValues(HashMap<ThreadLocal<Object>, Object> backup) {for (Map.Entry<ThreadLocal<Object>, Object> entry : backup.entrySet()) {final ThreadLocal<Object> threadLocal = entry.getKey();
        threadLocal.set(entry.getValue());
    }
}

四 ExecutorTtlWrapper

1 ExecutorTtlWrapper

ExecutorTtlWrapper 的代码非常少：

// com.alibaba.ttl.threadpool.ExecutorTtlWrapper
class ExecutorTtlWrapper implements Executor, TtlWrapper<Executor>, TtlEnhanced {
    
    // 这个变量代表了一个线程池
    private final Executor executor;
    // 这个变量是一个幂等标识符
    protected final boolean idempotent;

    ExecutorTtlWrapper(Executor executor, boolean idempotent) {
        this.executor = executor;
        this.idempotent = idempotent;
    }

    @Override
    public void execute(Runnable command) {executor.execute(TtlRunnable.get(command, false, idempotent));
    }

    @Overrid
    public Executor unwrap() {return executor;}

    // 其它办法不重要，这里省略...
}

ExecutorTtlWrapper 实质上是一个线程池的代理，在执行 execute(…) 办法的时候，会将 Runnable 工作包装成 TtlRunnable。

2 TtlEnhanced

// 这是一个单纯的空接口，用来标识一个类
public interface TtlEnhanced {}

3 TtlWrapper

// TtlWrapper 用来标识一个包装类
// 须要实现获取被包装对象的 unwrap 办法
public interface TtlWrapper<T> extends TtlEnhanced {T unwrap();
}

4 TtlExecutors

TtlExecutors 是一个动态工具类，用来生成 ExecutorTtlWrapper。

// com.alibaba.ttl.threadpool.TtlExecutors
public static Executor getTtlExecutor(Executor executor) {
    // 如果曾经包装过了，那么此处间接返回
    if (TtlAgent.isTtlAgentLoaded() || null == executor || executor instanceof TtlEnhanced) {return executor;}
    
    // 如果没有包装过，那么此处包装一下
    // 幂等标识符，此处默认为 true
    return new ExecutorTtlWrapper(executor, true);
}

TtlAgent 是对探针技术的利用，临时不开展解说。

五 TtlRunnable

1 TtlRunnable

首先来看一下 class：

// com.alibaba.ttl.TtlRunnable
public final class TtlRunnable implements Runnable, TtlWrapper<Runnable>, TtlEnhanced, TtlAttachments {
    
    private final AtomicReference<Object> capturedRef;
    private final Runnable runnable;
    private final boolean releaseTtlValueReferenceAfterRun;
    
    private TtlRunnable(Runnable runnable, boolean releaseTtlValueReferenceAfterRun) {// capture() 办法见下面 第三 part 的 Transmitter 局部
        // 实质上这是以后线程所存储的 TransmittableThreadLocal 和 ThreadLocal 的快照
        this.capturedRef = new AtomicReference<Object>(capture());
        // 实在的业务逻辑
        this.runnable = runnable;
        // 以后 TtlRunnable 是否能够反复执行
        // true 的状况下，只有执行完，就不能反复执行了
        this.releaseTtlValueReferenceAfterRun = releaseTtlValueReferenceAfterRun;
    }
    
    // 其它办法先省略...   
}

2 get

TtlRunnable.get(…) 是一个静态方法，用于创立一个 TtlRunnable 对象。

// com.alibaba.ttl.TtlRunnable
public static TtlRunnable get(Runnable runnable) {return get(runnable, false, false);
}

public static TtlRunnable get(Runnable runnable, boolean releaseTtlValueReferenceAfterRun) {return get(runnable, releaseTtlValueReferenceAfterRun, false);
}

public static TtlRunnable get(Runnable runnable, boolean releaseTtlValueReferenceAfterRun, boolean idempotent) {
    // 空判断
    if (null == runnable) 
        return null;

    // 如果以后为幂等，则此处复用
    if (runnable instanceof TtlEnhanced) {if (idempotent) 
            return (TtlRunnable) runnable;
        else 
            throw new IllegalStateException("Already TtlRunnable!");
    }
    
    // 创建对象
    return new TtlRunnable(runnable, releaseTtlValueReferenceAfterRun);
}

3 run

TtlRunnable.run() 是外围办法，是对业务逻辑的封装。

// com.alibaba.ttl.TtlRunnable
public void run() {
    
    // 获取以后快照
    final Object captured = capturedRef.get();
    
    // 有效性判断
    if (captured == null || releaseTtlValueReferenceAfterRun && !capturedRef.compareAndSet(captured, null)) {throw new IllegalStateException("TTL value reference is released after run!");
    }

    // replay 办法来自 Transmitter
    // 用于创立一个以后线程的 ThreadLocal 的备份
    final Object backup = replay(captured);
    try {runnable.run();
    } finally {
        // restore 办法来自 Transmitter
        // 应用备份来复原以后线程的 ThreadLocal 数据
        restore(backup);
    }
}

captured 实际上是一个备忘录模式，用于确保子线程内的数据批改不影响到父线程。

六 一点唠叨

• 封装的很有意思，然而很多细节还是没太看懂
• 仅为集体的学习笔记，可能存在谬误或者表述不清的中央，有缘补充