ThreadLocal及阿里（TransmittableThreadLocal，TTL）分析

news2024/11/13 16:21:27

TTL类关系图
ThreadLocal <- InheritableThreadLocal <- TransmittableThreadLocal

在这里插入图片描述

1. ThreadLocal

ThreadLocal 类提供线程本地（局部）变量。每个线程都有自己独立初始化的变量副本。
TheadLocal 允许我们存储仅由特定线程访问的数据，从而起到线程隔离的作用，避免了并发场景下的线程安全问题。

1.1 使用场景

线程持有自己的数据变量

/**
 * 每个线程拥有自己的Integer变量，默认初始化为0
 */
private static final ThreadLocal<Integer> INTEGER_THREAD_LOCAL = ThreadLocal.withInitial(() -> 0);

避免数据竞争，每个线程持有自己的线程变量

/**
 * 每个线程拥有自己的SDF，避免竞争，保证线程安全
 */
private static final ThreadLocal<SimpleDateFormat> SDF_TL = ThreadLocal.withInitial(() -> {
    return new SimpleDateFormat("yyyyMMddHHmmss");
});

数据传递，跨方法级别数据传递

/**
 * 数据传递上下文： LogbackMDCAdapter MDC 实际功能实现
 */
final ThreadLocal<Map<String, String>> context = new ThreadLocal<>();

1.2 ThreadLocal主要方法

// get() 获取线程本地变量，如果没有，则调用setInitialValue() 初始化并设置
public T get() {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null) {
        ThreadLocalMap.Entry e = map.getEntry(this);
        if (e != null) {
            @SuppressWarnings("unchecked")
            T result = (T)e.value;
            return result;
        }
    }
    return setInitialValue();
}
// 初始化
private T setInitialValue() {
    T value = initialValue();
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
    return value;
}
// 设置线程本地变量
public void set(T value) {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
}
// 移除线程本地变量
public void remove() {
    ThreadLocalMap m = getMap(Thread.currentThread());
    if (m != null)
        m.remove(this);
}

1.3 实现原理

ThreadLocal实现主要依赖
1.ThreadLocal实例对象，作为线程访问数据的KEY存在。
2.ThreadLocalMap对象，每个线程有自己的map，ThreadLocal实例作为KEY。
注意：此处entry中的ThreadLocal key使用弱引用，防止内存泄漏，在清除方法中会清除所有 key为null的entry。

static class ThreadLocalMap {
    static class Entry extends WeakReference<ThreadLocal<?>> {
        /** The value associated with this ThreadLocal. */
        Object value;
        Entry(ThreadLocal<?> k, Object v) {
            super(k);
            value = v;
        }
    }
}

ThreadLoclMap作为每个线程实例的字段存储在线程实例中：

public class Thread implements Runnable {
    /**
     * 与此线程相关的ThreadLocal值。此Map由 ThreadLocal 类维护。
     */
    ThreadLocal.ThreadLocalMap threadLocals = null;
    /**
     * 与此线程相关的 InheritableThreadLocal 值。此映射由 InheritableThreadLocal 类维护
     */
    ThreadLocal.ThreadLocalMap inheritableThreadLocals = null;
}

1.4 缺陷

ThreadLocal在创建子线程时不会把线程本地变量拷贝到子线程中，这就会导致子线程无法获取到本地线程保存的线程信息

2. InheritableThreadLocal

InheritableThreadLocal 可继承的ThreadLocal，继承并扩扩展了原始ThreadLocal的功能，提供从父线程到子线程的value继承：创建子线程时，子线程接收父线程具有的 inheritable thread-local 值。

2.1 使用场景

新建子线程需要集成父线程中的线程本地变量时，可以使用 InheritableThreadLocal 实现

/**
 * 可继承ThreadLocal，子线程将继承父线程中ITL变量value
 */
private static final ThreadLocal<Integer> BIZ_ITL = new InheritableThreadLocal();

/**
 * 运行结果：
 * main-thread-get:123
 * sub-thread-get:123
 */
public static void main(String[] args) {
    BIZ_ITL.set(123);
    System.out.println("main-thread-get:"+BIZ_ITL.get());
    new Thread(()->{
        System.out.println("sub-thread-get:"+BIZ_ITL.get());
    }).start();
}

2.2 InheritableThreadLocal 主要方法

getMap() createMap() 使用 ITLMap ，非 TLMap

ThreadLocalMap getMap(Thread t) {
   return t.inheritableThreadLocals;
}
void createMap(Thread t, T firstValue) {
    t.inheritableThreadLocals = new ThreadLocalMap(this, firstValue);
}

2.3 实现原理

main线程调用get方法跳转到InheritableThreadLocal重写的createMap，让inheritableThreadLocals赋初值；后续http请求会copy main线程中的数据，同样赋予inheritableThreadLocals 值。
子程继承父线程线程本地变量值是在Thread创建时，copy父类现成中的 ITLMap中的key和Value：
Thread构造函数：调用init方法，在init方法中子线程根据父线程的ITLMap创建自己的ITLMap（传递、继承）。

// 初始化Thread
public Thread() {
    init(null, null, "Thread-" + nextThreadNum(), 0);
}
private void init(ThreadGroup g, Runnable target, String name,
                      long stackSize, AccessControlContext acc) {
    // 获取父线程
    Thread parent = currentThread();
    // 省略其他步骤
    if (parent.inheritableThreadLocals != null)
            this.inheritableThreadLocals =
                ThreadLocal.createInheritedMap(parent.inheritableThreadLocals);
}

2.4 缺陷

使用线程池时，线程不会每次都创建，也就不会进行父子线程之间的数据传递；如果使用池化的线程，就不会进行父子线程本地变量的拷贝。

3. TransmittableThreadLocal

3.1 使用场景

ThreadLocal和InheritableThreadLocal 能够完成变量的线程本地化和父子线程中的value传递。
但是现实项目中大多数线程池化在线程池中，因此，提交任务的线程无法将提交现成的本地变量传递给执行task的任务线程。
TTL组件功能：在使用线程池等会池化复用线程的执行组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题。
业务期望：上下文生命周期的操作从业务逻辑中分离出来。业务逻辑不涉及生命周期，就不会有业务代码如疏忽清理而引发的问题了。

3.2 TransmittableThreadLocal主要方法

整个上下文的传递流程或说生命周期可以规范化成：
捕捉、回放和恢复这3个操作，即CRR(capture/replay/restore)模式。
CRR :

capture方法：抓取线程（线程A）的所有TTL值。
replay方法：在另一个线程（线程B）中，回放在capture方法中抓取的TTL值，并返回回放前TTL值的备份
restore方法：恢复线程B执行replay方法之前的TTL值（即备份）

// 实现代码示例 TtlRunnable.class
// 初始化TtlRunnable对象
private TtlRunnable(@NonNull Runnable runnable, boolean releaseTtlValueReferenceAfterRun) {
	// TransmittableThreadLocal.Transmitter.capture() 抓取当前线程中的TTL值的备份
    this.capturedRef = new AtomicReference<>(capture());
    this.runnable = runnable;
    this.releaseTtlValueReferenceAfterRun = releaseTtlValueReferenceAfterRun;
}

@Override
public void run() {
	// 取出TTL值备份数据
    final Object captured = capturedRef.get();
    if (captured == null || releaseTtlValueReferenceAfterRun && !capturedRef.compareAndSet(captured, null)) {
        throw new IllegalStateException("TTL value reference is released after run!");
    }
    // 回放在capture方法中抓取的TTL值并返回回放前TTL值的备份
    final Object backup = replay(captured);
    try {
        runnable.run();
    } finally {
        // 恢复线程B执行replay方法之前的TTL值（即备份）
        restore(backup);
    }
}

3.3 使用样例

private static final int DEFAULT_SIZE = 8;
private static final TransmittableThreadLocal<Map<String, Object>> CACHE = new TransmittableThreadLocal<Map<String, Object>>() {
    @Override
    protected Map<String, Object> initialValue() {
        return new HashMap<>(DEFAULT_SIZE);
    }
};

Executor ttlExecutor = TtlExecutors.getTtlExecutor(new ThreadPoolExecutor(
       properties.getCoreSize(),
        properties.getMaxSize(),
        properties.getKeepAliveSeconds(),
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(),
        new ThreadPoolExecutor.CallerRunsPolicy()
));

ttlExecutor.execute(TtlRunnable.get(() -> {}));