TL_ITL_TTL原理

GPT摘要

该文章讨论了线程本地存储（ThreadLocal）及其扩展机制，包括InheritableThreadLocal和TransmittableThreadLocal（TTL）的原理与实现，以及它们的应用场景和局限性。以下是核心内容的总结： 1. ThreadLocal基础 - ThreadLocal为每个线程分配独立的空间（ThreadLocalMap），存储线程私有变量，确保线程隔离。 - 用途：存储线程链路中的用户信息（如请求上下文），线程间互不干扰。 - 底层通过弱引用（WeakReference）的Entry数组实现，哈希冲突时使用线性探测。 2. 父子线程传递问题 - 普通ThreadLocal的局限：子线程无法继承父线程的ThreadLocal变量（子线程读取为null）。 - InheritableThreadLocal的改进： - 通过inheritableThreadLocals空间在子线程创建时复制父线程的Entry，实现继承。 - 但仅支持线程创建时的静态传递，无法应对线程池复用线程的场景（值不会更新）。 3. 线程池场景的解决方案——TransmittableThreadLocal（TTL） - 核心思想：在执行任务时动态拷贝父线程的TTL变量（而非线程创建时）。 - 实现机制： - 通过TtlRunnable装饰器在任务执行前备份父线程的TTL值，执行后恢复。 - holder（InheritableThreadLocal集合）记录所有需传递的TTL变量，通过capture、replay和restore完成值传递与还原。 - 优势：支持线程池等动态线程复用场景，确保任务执行时能获取最新的父线程变量。 4. 应用场景 - 分布式跟踪：传递请求链路标识（如TraceID）。 - 日志上下文：跨线程记录用户会话信息。 - Session级缓存：线程安全的用户数据存储。 - 跨框架/SDK通信：上层向下层透传上下文（如租户ID）。 5. …

ThreadLocal

  • 为每一个线程开辟了一个空间Map，用于存储变量
  • 用额外的空间使得线程之间互不影响
  • 用途：在一个线程中，存储用户的信息，使得整个线程链路都可以访问到，并且线程之间互不影响
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public void set(T value) {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t); 
    // 当前线程的一个变量t.threadLocals; 本质上是Entry[] table，key是tl的弱引用，值是value
    // index 是int i = key.threadLocalHashCode & (table.length - 1); 冲突了就线性探测
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
}

static class Entry extends WeakReference<ThreadLocal<?>> {
    /** The value associated with this ThreadLocal. */
    Object value;

    Entry(ThreadLocal<?> k, Object v) {
        super(k);
        value = v;
    }
}

问题：当创建子线程时，子线程无法读取到

1
2
3
4
5
new Thread(() -> {
log.info(Thread.currentThread() + "in thread: " + tl.get());
}).start();

2024-08-12 16:19:53 INFO  [Thread-2] c.e.t.ThreadLocalStudyApplicationTests - Thread[Thread-2,5,main]in thread: null

InheritableThreadLocal

  • 实现了父子线程间的传递

  • 重写了getMap()方法，返回的是t.inheritableThreadLocals，相当于再开辟了一个空间

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    protected T childValue(T parentValue) { // childValue是复制，父子会相互影响 想要不影响重写这个方法
        return parentValue;
    }
      
    ThreadLocalMap getMap(Thread t) {
       return t.inheritableThreadLocals;
    }
      
    void createMap(Thread t, T firstValue) {
        t.inheritableThreadLocals = new ThreadLocalMap(this, firstValue);
    }

  • 父传递给子在thread的创建时，将父的inheritableThreadLocals中的Entry复制给子

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    if (inheritThreadLocals && parent.inheritableThreadLocals != null)
                this.inheritableThreadLocals =
                    ThreadLocal.createInheritedMap(parent.inheritableThreadLocals);

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

        for (int j = 0; j < len; j++) {
            Entry e = parentTable[j];
            if (e != null) {
                @SuppressWarnings("unchecked")
                ThreadLocal<Object> key = (ThreadLocal<Object>) e.get();
                if (key != null) {
                    Object value = key.childValue(e.value); // childValue是复制，会相互影响 想要不影响重写这个方法
                    Entry c = new Entry(key, value);
                    int h = key.threadLocalHashCode & (len - 1);
                    while (table[h] != null)
                        h = nextIndex(h, len);
                    table[h] = c;
                    size++;
                }
            }
        }
    }

问题：传递过程只在线程创建时，如果是线程池这种已经创建好的，值就不会传递

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
ThreadLocal<String> tl = new InheritableThreadLocal<>();
tl.set("value1");

log.info("in main: " + tl.get());

ExecutorService service = Executors.newFixedThreadPool(3);

for (int i = 0; i < 5; i++) {
    service.submit(() -> {
        log.info("in pool: " + tl.get());
    });
}
tl.set("value2");

for (int i = 0; i < 5; i++) {
    service.submit(() -> {
        log.info("in pool: " + tl.get()); // 值还是value1 不会修改
    });
}

TransmittableThreadLocal

  • 基本思想：ITL通过在创建时拷贝数据实现传递，更进一步，在执行时提前进行一次拷贝即可

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    public class TransmittableThreadLocal<T> extends InheritableThreadLocal<T> implements TtlCopier<T> {
        // 使用一个InheritableThreadLocal的Map，记录下全部的TTL，在set时TTL会被添加进holder;
        private static final InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>> holder = new InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>>(){}
    }


    public final void set(T value) {
        if (!this.disableIgnoreNullValueSemantics && null == value) {
            this.remove();
        } else {
            super.set(value);
            this.addThisToHolder();
        }

    }

    private void addThisToHolder() {
        if (!((WeakHashMap)holder.get()).containsKey(this)) {
            ((WeakHashMap)holder.get()).put(this, (Object)null);
        }

    }

  • TtlRunnable对runable进行一层包裹（装饰器模式），对TTL变量进行备份 赋值 以及恢复

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    public void run() {
        Object captured = this.capturedRef.get();
        if (captured != null && (!this.releaseTtlValueReferenceAfterRun || this.capturedRef.compareAndSet(captured, (Object)null))) {
            Object backup = Transmitter.replay(captured);

            try {
                this.runnable.run();
            } finally {
                Transmitter.restore(backup);
            }

        } else {
            throw new IllegalStateException("TTL value reference is released after run!");
        }
    }
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    // 获取所有父的ttl
    private static HashMap<TransmittableThreadLocal<Object>, Object> captureTtlValues() {
        HashMap<TransmittableThreadLocal<Object>, Object> ttl2Value = new HashMap();
        Iterator var1 = ((WeakHashMap)TransmittableThreadLocal.holder.get()).keySet().iterator();

        while(var1.hasNext()) {
            TransmittableThreadLocal<Object> threadLocal = (TransmittableThreadLocal)var1.next();
            ttl2Value.put(threadLocal, threadLocal.copyValue());
        }

        return ttl2Value;
    }

    // 备份并替换为父的
    private static HashMap<TransmittableThreadLocal<Object>, Object> replayTtlValues(@NonNull HashMap<TransmittableThreadLocal<Object>, Object> captured) {
        HashMap<TransmittableThreadLocal<Object>, Object> backup = new HashMap();
        Iterator<TransmittableThreadLocal<Object>> iterator = ((WeakHashMap)TransmittableThreadLocal.holder.get()).keySet().iterator();

        while(iterator.hasNext()) {
            TransmittableThreadLocal<Object> threadLocal = (TransmittableThreadLocal)iterator.next();
            backup.put(threadLocal, threadLocal.get());
            if (!captured.containsKey(threadLocal)) {
                iterator.remove();
                threadLocal.superRemove();
            }
        }

        setTtlValuesTo(captured); // 设置值为父的
        TransmittableThreadLocal.doExecuteCallback(true);
        return backup;
    }

    private static void setTtlValuesTo(@NonNull HashMap<TransmittableThreadLocal<Object>, Object> ttlValues) {
        Iterator var1 = ttlValues.entrySet().iterator();

        while(var1.hasNext()) {
            Map.Entry<TransmittableThreadLocal<Object>, Object> entry = (Map.Entry)var1.next();
            TransmittableThreadLocal<Object> threadLocal = (TransmittableThreadLocal)entry.getKey();
            threadLocal.set(entry.getValue());
        }
    }

至此可以实现父线程传递给子线程信息，可以用来

  • 分布式跟踪系统
  • 日志收集记录系统上下文
  • Session级Cache
  • 应用容器或上层框架跨应用代码给下层SDK传递信息