阿里TransmittableThreadLocal实践

字数统计: 2.6k字   |   阅读时长≈ 13分

前言

       Java分布式系统中经常会用到 ThreadLocal 来解决线程间数据隔离性的问题，但是 ThreadLocal 存在父子线程间值无法传递的局限性，为了解决上面的问题 InheritableThreadLocal 就应运而生，然而 InheritableThreadLocal 又存在着池化子线程与父线程间值无法传递的局限性，为了解决上面这些问题，最终阿里的 TransmittableThreadLocal 脱颖而出，本文将重点以代码来演示TTL的一些用法。

一、TransmittableThreadLocal简介

在使用线程池等会池化复用线程的执行组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题。

JDK的InheritableThreadLocal类可以完成父线程到子线程的值传递。但对于使用线程池等会池化复用线程的执行组件的情况，线程由线程池创建好，并且线程是池化起来反复使用的；这时父子线程关系的ThreadLocal值传递已经没有意义，应用需要的实际上是把 任务提交给线程池时的ThreadLocal值传递到 任务执行时。

本库提供的TransmittableThreadLocal类继承并加强InheritableThreadLocal类，解决上述的问题，使用详见User Guide。

二、TransmittableThreadLocal用法

1. 简单使用

父线程给子线程传递值。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();

// =====================================================

// 在父线程中设置
context.set("value-set-in-parent");

// =====================================================

// 在子线程中可以读取，值是"value-set-in-parent"
String value = context.get();

# 完整可运行的Demo代码参见SimpleDemo.kt。

这是其实是InheritableThreadLocal的功能，应该使用InheritableThreadLocal来完成。

但对于使用线程池等会池化复用线程的执行组件的情况，线程由线程池创建好，并且线程是池化起来反复使用的；这时父子线程关系的ThreadLocal值传递已经没有意义，应用需要的实际上是把 任务提交给线程池时 的ThreadLocal值传递到 任务执行时。

解决方法参见下面的这几种用法。

2. 保证线程池中传递值

2.1 修饰Runnable和Callable

使用TtlRunnable和TtlCallable来修饰传入线程池的Runnable和Callable。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();

// =====================================================

// 在父线程中设置
context.set("value-set-in-parent");

Runnable task = new RunnableTask();
// 额外的处理，生成修饰了的对象ttlRunnable
Runnable ttlRunnable = TtlRunnable.get(task);
executorService.submit(ttlRunnable);

// =====================================================

// Task中可以读取，值是"value-set-in-parent"
String value = context.get();

上面演示了Runnable，Callable的处理类似

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();

// =====================================================

// 在父线程中设置
context.set("value-set-in-parent");

Callable call = new CallableTask();
// 额外的处理，生成修饰了的对象ttlCallable
Callable ttlCallable = TtlCallable.get(call);
executorService.submit(ttlCallable);

// =====================================================

// Call中可以读取，值是"value-set-in-parent"
String value = context.get();

# 完整可运行的Demo代码参见TtlWrapperDemo.kt。

整个过程的完整时序图

时序图

2.2 修饰线程池

省去每次Runnable和Callable传入线程池时的修饰，这个逻辑可以在线程池中完成。

通过工具类com.alibaba.ttl.threadpool.TtlExecutors完成，有下面的方法：

  • getTtlExecutor：修饰接口Executor
  • getTtlExecutorService：修饰接口ExecutorService
  • getTtlScheduledExecutorService：修饰接口ScheduledExecutorService

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
ExecutorService executorService = ...
// 额外的处理，生成修饰了的对象executorService
executorService = TtlExecutors.getTtlExecutorService(executorService);

TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();

// =====================================================

// 在父线程中设置
context.set("value-set-in-parent");

Runnable task = new RunnableTask();
Callable call = new CallableTask();
executorService.submit(task);
executorService.submit(call);

// =====================================================

// Task或是Call中可以读取，值是"value-set-in-parent"
String value = context.get();

# 完整可运行的Demo代码参见TtlExecutorWrapperDemo.kt。

2.3 使用Java Agent来修饰JDK线程池实现类

这种方式，实现线程池的传递是透明的，业务代码中没有修饰Runnable或是线程池的代码。即可以做到应用代码 无侵入。
# 关于 无侵入 的更多说明参见文档Java Agent方式对应用代码无侵入。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// ## 1. 框架上层逻辑，后续流程框架调用业务 ##
TransmittableThreadLocal<String> context = new TransmittableThreadLocal<>();
context.set("value-set-in-parent");

// ## 2. 应用逻辑，后续流程业务调用框架下层逻辑 ##
ExecutorService executorService = Executors.newFixedThreadPool(3);

Runnable task = new RunnableTask();
Callable call = new CallableTask();
executorService.submit(task);
executorService.submit(call);

// ## 3. 框架下层逻辑 ##
// Task或是Call中可以读取，值是"value-set-in-parent"
String value = context.get();

Demo参见AgentDemo.kt。执行工程下的脚本scripts/run-agent-demo.sh即可运行Demo。

三、TransmittableThreadLocal实践

测试类代码
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
/**
 * @author yinshuang
 * @date 2023/8/19 9:26
 * @description ThreadLocalTest
 */
@Slf4j
public class ThreadLocalTest {

    private static ExecutorService executorService;

    ThreadLocal<String> tlStr = new ThreadLocal<>();
    InheritableThreadLocal<String> itlStr = new InheritableThreadLocal<>();
    TransmittableThreadLocal<String> ttlStr = new TransmittableThreadLocal<>();

    TransmittableThreadLocal<Map<String, Object>> ttlMap1 = new TransmittableThreadLocal<Map<String, Object>>() {
        @Override
        protected Map<String, Object> initialValue() {
            log.info("init ttl map");
            return new LinkedHashMap<>();
        }
    };

    TransmittableThreadLocal<Map<String, Object>> ttlMap2 = new TransmittableThreadLocal<Map<String, Object>>() {
        @Override
        protected Map<String, Object> initialValue() {
            log.info("init ttl map");
            return new LinkedHashMap<>();
        }
        @Override
        public Map<String, Object> copy(Map<String, Object> parentValue) {
            log.info("copy ttl map");
            return new LinkedHashMap<>(parentValue);
        }
    };

    @Test
    @DisplayName("ThreadLocal<String>变量演示")
    public void testThreadLocalStr() {
        if (executorService == null) {
            ExecutorService executor = new ThreadPoolExecutor(3, 3, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));
            executorService = TtlExecutors.getTtlExecutorService(executor);
            log.info("创建ttl线程池!");
        }
        IntStream.range(0, 3).forEach(this::threadLocalStr);
        if (executorService != null) {
            executorService.shutdown();
            executorService = null;
            log.info("销毁ttl线程池!");
        }
    }

    private void threadLocalStr(Integer i) {
        Thread t1 = Thread.currentThread();
        tlStr.set(String.valueOf(i));
        itlStr.set(String.valueOf(i));
        ttlStr.set(String.valueOf(i));
        log.info("ThreadLocal in [{}]: {}", t1.getName(), tlStr.get());
        log.info("InheritableThreadLocal in [{}]: {}", t1.getName(), itlStr.get());
        log.info("TransmittableThreadLocal in [{}]: {}", t1.getName(),ttlStr.get());
        executorService.execute(() -> {
            Thread t2 = Thread.currentThread();
            log.info("ThreadLocal in [{}]: {}", t2.getName(), tlStr.get());
            log.info("InheritableThreadLocal in [{}]: {}", t2.getName(), itlStr.get());
            log.info("TransmittableThreadLocal in [{}]: {}", t2.getName(), ttlStr.get());
        });
        tlStr.remove();
        itlStr.remove();
        ttlStr.remove();
    }

    @Test
    @DisplayName("ThreadLocal<Map>变量演示, 父子线程Map共享和父子线程Map独立")
    public void testThreadLocalMap() {
        if (executorService == null) {
            ExecutorService executor = new ThreadPoolExecutor(3, 3, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));
            executorService = TtlExecutors.getTtlExecutorService(executor);
            log.info("创建ttl线程池!");
        }
        IntStream.range(0, 3).forEach(this::threadLocalMap);
        if (executorService != null) {
            executorService.shutdown();
            executorService = null;
            log.info("销毁ttl线程池!");
        }
    }

    private void threadLocalMap(Integer i) {
        Thread t1 = Thread.currentThread();
        ttlMap1.get().put(t1.getName(), i);
        ttlMap2.get().put(t1.getName(), i);
        log.info("TransmittableThreadLocal1 get in [{}] change before: {}", t1.getName(), ttlMap1.get());
        log.info("TransmittableThreadLocal2 get in [{}] change before: {}", t1.getName(), ttlMap2.get());
        assert executorService != null;
        executorService.execute(() -> {
            try {
                // 滞后多线程打印ttlMap，演示父子线程Map变量的影响
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            Thread t2 = Thread.currentThread();
            log.info("TransmittableThreadLocal1 in [{}]: {}", t2.getName(), ttlMap1.get());
            log.info("TransmittableThreadLocal2 in [{}]: {}", t2.getName(), ttlMap2.get());
        });
        ttlMap1.get().put(t1.getName(), i*10+5);
        ttlMap2.get().put(t1.getName(), i*10+5);
        log.info("TransmittableThreadLocal1 get in [{}] change after: {}", t1.getName(), ttlMap1.get());
        log.info("TransmittableThreadLocal2 get in [{}] change after: {}", t1.getName(), ttlMap2.get());
        ttlMap1.remove();
        ttlMap2.remove();
    }
}

threadLocalStr运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
09:13:09.290 [main] INFO com.chinacscs.mp.ThreadLocalTest - 创建ttl线程池!
09:13:09.296 [main] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [main]: 0
09:13:09.298 [main] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [main]: 0
09:13:09.298 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [main]: 0
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [main]: 1
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [main]: 1
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [main]: 1
09:13:09.307 [pool-1-thread-2] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [pool-1-thread-2]: null
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [main]: 2
09:13:09.307 [pool-1-thread-2] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [pool-1-thread-2]: null
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [main]: 2
09:13:09.307 [pool-1-thread-2] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [pool-1-thread-2]: 0
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [main]: 2
09:13:09.307 [pool-1-thread-1] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [pool-1-thread-1]: null
09:13:09.307 [pool-1-thread-1] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [pool-1-thread-1]: null
09:13:09.307 [pool-1-thread-1] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [pool-1-thread-1]: 1
09:13:09.307 [main] INFO com.chinacscs.mp.ThreadLocalTest - 销毁ttl线程池!
09:13:09.307 [pool-1-thread-3] INFO com.chinacscs.mp.ThreadLocalTest - ThreadLocal in [pool-1-thread-3]: null
09:13:09.308 [pool-1-thread-3] INFO com.chinacscs.mp.ThreadLocalTest - InheritableThreadLocal in [pool-1-thread-3]: null
09:13:09.308 [pool-1-thread-3] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal in [pool-1-thread-3]: 2

从执行结果可以看出 ThreadLocal 和 InheritableThreadLocal 都无法在池化线程子线程获取父线程的值

threadLocalMap运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
09:14:19.330 [main] INFO com.chinacscs.mp.ThreadLocalTest - 创建ttl线程池!
09:14:19.334 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.334 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.334 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change before: {main=0}
09:14:19.336 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change before: {main=0}
09:14:19.342 [main] INFO com.chinacscs.mp.ThreadLocalTest - copy ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change after: {main=5}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change after: {main=5}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change before: {main=1}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change before: {main=1}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - copy ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change after: {main=15}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change after: {main=15}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - init ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change before: {main=2}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change before: {main=2}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - copy ttl map
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal1 get in [main] change after: {main=25}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - TransmittableThreadLocal2 get in [main] change after: {main=25}
09:14:19.343 [main] INFO com.chinacscs.mp.ThreadLocalTest - 销毁ttl线程池!

从执行结果可以看出，TTL默认创建的Map变量在父子线程会相互影响，两者用的是同一个Map，查看源码TTL线程默认的copy方法复制的是父线程Map的引用，所以父线程的改动会影响子线程的Map一起变化，重写copy方法新建一个Map后父线程和子线程就不会相互影响了

参考文章

1.ThreadLocal系列之——父子线程传递线程私有数据
2.ThreadLocal父子线程传递实现方案
3.阿里开源TransmittableThreadLocal使用经验记录
4.TransmittableThreadLocal源码分析

赏

谢谢你请我吃糖果

微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置：

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发


联系QQ:875038467

很惭愧

只做了一点微小的工作
谢谢大家