1、线上问题

提醒单手动发送短信发送不了，没有日志定位

核心代码：

img

img

img

2、问题猜想

  1. 线上复现发送异常, 查询到对应时间日志, 消息配置为空, 现实是消息配置没有问题
  2. 线上发送偶尔有问题, 偶尔正常, pre 则完全正常
  3. 查看日志合理怀疑, 异步时, 获取 groupId 异常, F6static 使用了 TransmittableThreadLocal 获取
  4. 按正常道理来说, 使用 TransmittableThreadLocal 应该不存在这个问题, 查看 TransmittableThreadLocal 文档, 使用线程池对象需要使用 TtlExecutors.getTtlExecutorService 修饰
  5. 怀疑 @Async 注解, 有默认线程池对象, 当时写代码, 没有添加 async 单独使用的线程池, 所以更没有将 异步线程池使用 TtlExecutors 修饰

3、验证猜想

3.1、验证线程池修饰问题

3.1.1、未修饰程池, 导致 ttl 获取上下文异常

  • 添加测试文件, 再两个主线程内, 添加普通线程池异步, 线程池对象没有使用 TtlExecutors.getTtlExecutorService 修饰
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
public class NormalExecutorsTest {
    private static ExecutorService executorService = Executors.newFixedThreadPool(2);
    private static ThreadLocal tl = new TransmittableThreadLocal<>(); //这里采用TTL的实现
    public static void main(String[] args) {
        new Thread(() -> {
            String mainThreadName = "main_01";
            tl.set(1);
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(2); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));
        }).start();

        new Thread(() -> {
            String mainThreadName = "main_02";
            tl.set(3);
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(4); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));
        }).start();

    }

    private static void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

img

从上述测试发现, 两个主线程里都使用线程池异步，而且值在主线程里还发生过改变，测试结果显示确实获取上下文异常，发生错乱

3.1.2、修饰线程池, 获取上下文信息正常

添加测试文件, 区别只有线程池使用了 TtlExecutors.getTtlExecutorService 修饰

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
public class TtlExecutorsTest {

    // 需要注意的是，使用TTL的时候，要想传递的值不出问题，线程池必须得用TTL加一层代理（下面会讲这样做的目的）
    private static ExecutorService executorService = TtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(2));
    private static ThreadLocal tl = new TransmittableThreadLocal<>(); //这里采用TTL的实现
    public static void main(String[] args) {
        new Thread(() -> {
            String mainThreadName = "main_01";
            tl.set(1);
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(2); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));
        }).start();


        new Thread(() -> {
            String mainThreadName = "main_02";
            tl.set(3);
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(4); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            executorService.execute(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            });
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));
        }).start();
    }

    private static void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

img

不难发现，两个主线程里都使用线程池异步，而且值在主线程里还发生过改变，测试结果展示一切正常，由此可以知道 TTL 在使用线程池的情况下，只要使用 TtlExecutors 对应方法修饰线程池, 也可以很好的完成传递，而且不会发生错乱。

3.1.3、普通异步请求不使用线程池, 正常获取上下文信息

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
public class NormalAsyncTest {
   
    private static ThreadLocal tl = new TransmittableThreadLocal<>(); //这里采用TTL的实现
    public static void main(String[] args) {
        new Thread(() -> {
            String mainThreadName = "main_01";
            tl.set(1);
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(1), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(2); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(2), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));

        }).start();

        new Thread(() -> {
            String mainThreadName = "main_02";
            tl.set(3);
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之前(3), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            sleep(1L); //确保上面的会在tl.set执行之前执行
            tl.set(4); // 等上面的线程池第一次启用完了，父线程再给自己赋值
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            new Thread(() -> {
                sleep(1L);
                System.out.println(String.format("本地变量改变之后(4), 父线程名称-%s, 子线程名称-%s, 变量值=%s", mainThreadName, Thread.currentThread().getName(), tl.get()));
            }).start();
            System.out.println(String.format("线程名称-%s, 变量值=%s", Thread.currentThread().getName(), tl.get()));
        }).start();

    }

    private static void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

img

测试结果显示和使用了线程池修饰的测试保持一致, 普通线程异步直接获取 ttl 上下文信息没有问题

3.2、验证 @Async 使用了默认线程池

3.2.1、断点异步方法, 查看线程池相关

img

很明显可以看到默认线程池配置

4、源码解析

4.1、TransmittableThreadLocal

img

接入文档显示的很清楚, 使用线程池需要 TtlExecutors 对应修饰

先来看 TTL 里面的几个重要属性及方法

TTL 定义：

1
public class TransmittableThreadLocal extends InheritableThreadLocal

可以看到，TTL 继承了 ITL，意味着 TTL 首先具备 ITL 的功能。

再来看看一个重要属性 holder：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/**
  * 这是一个ITL类型的对象，持有一个全局的WeakMap（weakMap的key是弱引用，同TL一样，也是为了解决内存泄漏的问题），里面存放了TTL对象
  * 并且重写了initialValue和childValue方法，尤其是childValue，可以看到在即将异步时父线程的属性是直接作为初始化值赋值给子线程的本地变量对象（TTL）的
  */
 private static InheritableThreadLocal<Map<TransmittableThreadLocal<?>, ?>> holder =
         new InheritableThreadLocal<Map<TransmittableThreadLocal<?>, ?>>() {
             @Override
             protected Map<TransmittableThreadLocal<?>, ?> initialValue() {
                 return new WeakHashMap<TransmittableThreadLocal<?>, Object>();
             }

             @Override
             protected Map<TransmittableThreadLocal<?>, ?> childValue(Map<TransmittableThreadLocal<?>, ?> parentValue) {
                 return new WeakHashMap<TransmittableThreadLocal<?>, Object>(parentValue);
             }
         };

再来看下 set 和 get：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
//下面的方法均属于TTL类
@Override
    public final void set(T value) {
        super.set(value);
        if (null == value) removeValue();
        else addValue();
    }

    @Override
    public final T get() {
        T value = super.get();
        if (null != value) addValue();
        return value;
    }
    
    private void removeValue() {
        holder.get().remove(this); //从holder持有的map对象中移除
    }

    private void addValue() {
        if (!holder.get().containsKey(this)) {
            holder.get().put(this, null); //从holder持有的map对象中添加
        }
    }

TTL 里先了解上述的几个方法及对象，可以看出，单纯的使用 TTL 是达不到支持线程池本地变量的传递的，通过第一部分的例子，可以发现，除了要启用 TTL，还需要通过 TtlExecutors.getTtlExecutorService 包装一下线程池才可以，那么，下面就来看看在程序即将通过线程池异步的时候，TTL 帮我们做了哪些操作（这一部分是 TTL 支持线程池传递的核心部分）：

首先打开包装类，看下 execute 方法在执行时做了些什么。

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
// 此方法属于线程池包装类ExecutorTtlWrapper
@Override
    public void execute(@Nonnull Runnable command) {
        executor.execute(TtlRunnable.get(command)); //这里会把Rannable包装一层，这是关键，有些逻辑处理，需要在run之前执行
    }

    // 对应上面的get方法，返回一个TtlRunnable对象，属于TtLRannable包装类
    @Nullable
    public static TtlRunnable get(@Nullable Runnable runnable) {
        return get(runnable, false, false);
    }

    // 对应上面的get方法
    @Nullable
    public static TtlRunnable get(@Nullable Runnable runnable, boolean releaseTtlValueReferenceAfterRun, boolean idempotent) {
        if (null == runnable) return null;

        if (runnable instanceof TtlEnhanced) { // 若发现已经是目标类型了（说明已经被包装过了）直接返回
            // avoid redundant decoration, and ensure idempotency
            if (idempotent) return (TtlRunnable) runnable;
            else throw new IllegalStateException("Already TtlRunnable!");
        }
        return new TtlRunnable(runnable, releaseTtlValueReferenceAfterRun); //最终初始化
    }

    // 对应上面的TtlRunnable方法
    private TtlRunnable(@Nonnull Runnable runnable, boolean releaseTtlValueReferenceAfterRun) {
        this.capturedRef = new AtomicReference<Object>(capture()); //这里将捕获后的父线程本地变量存储在当前对象的capturedRef里
        this.runnable = runnable;
        this.releaseTtlValueReferenceAfterRun = releaseTtlValueReferenceAfterRun;
    }

    // 对应上面的capture方法，用于捕获当前线程（父线程）里的本地变量，此方法属于TTL的静态内部类Transmitter
    @Nonnull
    public static Object capture() {
        Map<TransmittableThreadLocal<?>, Object> captured = new HashMap<TransmittableThreadLocal<?>, Object>();
        for (TransmittableThreadLocal<?> threadLocal : holder.get().keySet()) { // holder里目前存放的k-v里的key，就是需要传给子线程的TTL对象
            captured.put(threadLocal, threadLocal.copyValue());
        }
        return captured; // 这里返回的这个对象，就是当前将要使用线程池异步出来的子线程，所继承的本地变量合集
    }

    // 对应上面的copyValue，简单的将TTL对象里的值返回（结合之前的源码可以知道get方法其实就是获取当前线程（父线程）里的值，调用super.get方法）
    private T copyValue() {
        return copy(get());
    }
    protected T copy(T parentValue) {
        return parentValue;
    }

结合上述代码，大致知道了在线程池异步之前需要做的事情，其实就是把当前父线程里的本地变量取出来，然后赋值给 Rannable 包装类里的 capturedRef 属性，到此为止，下面会发生什么，我们大致上可以猜出来了，接下来大概率会在 run 方法里，将这些捕获到的值赋给子线程的 holder 赋对应的 TTL 值，那么我们继续往下看 Rannable 包装类里的 run 方法是怎么实现的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
//run方法属于Rannable的包装类TtlRunnable

@Override
    public void run() {
        Object captured = capturedRef.get(); // 获取由之前捕获到的父线程变量集
        if (captured == null || releaseTtlValueReferenceAfterRun && !capturedRef.compareAndSet(captured, null)) {
            throw new IllegalStateException("TTL value reference is released after run!");
        }

        /**
         * 重点方法replay，此方法用来给当前子线程赋本地变量，返回的backup是此子线程原来就有的本地变量值（原生本地变量，下面会详细讲），
         * backup用于恢复数据（如果任务执行完毕，意味着该子线程会归还线程池，那么需要将其原生本地变量属性恢复）
         */
        Object backup = replay(captured);
        try {
            runnable.run(); // 执行异步逻辑
        } finally {
            restore(backup); // 结合上面对于replay的解释，不难理解，这个方法就是用来恢复原有值的
        }
    }

根据上述代码，我们看到了 TTL 在异步任务执行前，会先进行赋值操作（就是拿着异步发生时捕获到的父线程的本地变量，赋给自己），当任务执行完，就恢复原生的自己本身的线程变量值。

下面来具体看这俩方法：

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
//下面的方法均属于TTL的静态内部类Transmittable

@Nonnull
    public static Object replay(@Nonnull Object captured) {
        @SuppressWarnings("unchecked")
        Map<TransmittableThreadLocal<?>, Object> capturedMap = (Map<TransmittableThreadLocal<?>, Object>) captured; //使用此线程异步时捕获到的父线程里的本地变量值
        Map<TransmittableThreadLocal<?>, Object> backup = new HashMap<TransmittableThreadLocal<?>, Object>(); //当前线程原生的本地变量，用于使用完线程后恢复用

        //注意：这里循环的是当前子线程原生的本地变量集合，与本方法相反，restore方法里循环这个holder是指：该线程运行期间产生的变量+父线程继承来的变量
        for (Iterator<? extends Map.Entry<TransmittableThreadLocal<?>, ?>> iterator = holder.get().entrySet().iterator();
             iterator.hasNext(); ) {
            Map.Entry<TransmittableThreadLocal<?>, ?> next = iterator.next();
            TransmittableThreadLocal<?> threadLocal = next.getKey();

            backup.put(threadLocal, threadLocal.get()); // 所有原生的本地变量都暂时存储在backup里，用于之后恢复用

            /**
             * 检查，如果捕获到的线程变量里，不包含当前原生变量值，则从当前原生变量里清除掉，对应的线程本地变量也清掉
             * 这就是为什么会将原生变量保存在backup里的原因，为了恢复原生值使用
             * 那么，为什么这里要清除掉呢？因为从使用这个子线程做异步那里，捕获到的本地变量并不包含原生的变量，当前线程
             * 在做任务时的首要目标，是将父线程里的变量完全传递给任务，如果不清除这个子线程原生的本地变量，
             * 意味着很可能会影响到任务里取值的准确性。
             *
             * 打个比方，有ttl对象tl，这个tl在线程池的某个子线程里存在对应的值2，当某个主线程使用该子线程做异步任务时
             * tl这个对象在当前主线程里没有值，那么如果不进行下面这一步的操作，那么在使用该子线程做的任务里就可以通过
             * 该tl对象取到值2，不符合预期
             */
            if (!capturedMap.containsKey(threadLocal)) {
                iterator.remove();
                threadLocal.superRemove();
            }
        }

        // 这一步就是直接把父线程本地变量赋值给当前线程了（这一步起就刷新了holder里的值了，具体往下看该方法，在异步线程运行期间，还可能产生别的本地变量，比如在真正的run方法内的业务代码，再用一个tl对象设置一个值）
        setTtlValuesTo(capturedMap);

        // 这个方法属于扩展方法，ttl本身支持重写异步任务执行前后的操作，这里不再具体赘述
        doExecuteCallback(true);

        return backup;
    }

    // 结合之前Rannable包装类的run方法来看，这个方法就是使用上面replay记录下的原生线程变量做恢复用的
    public static void restore(@Nonnull Object backup) {
        @SuppressWarnings("unchecked")
        Map<TransmittableThreadLocal<?>, Object> backupMap = (Map<TransmittableThreadLocal<?>, Object>) backup;
        // call afterExecute callback
        doExecuteCallback(false);

        // 注意，这里的holder取出来的，实际上是replay方法设置进去的关于父线程里的所有变量（结合上面来看，就是：该线程运行期间产生的变量+父线程继承来的变量）
        for (Iterator<? extends Map.Entry<TransmittableThreadLocal<?>, ?>> iterator = holder.get().entrySet().iterator();
             iterator.hasNext(); ) {
            Map.Entry<TransmittableThreadLocal<?>, ?> next = iterator.next();
            TransmittableThreadLocal<?> threadLocal = next.getKey();

            /**
             * 同样的，如果子线程原生变量不包含某个父线程传来的对象，那么就删除，可以思考下，这里的清除跟上面replay里的有什么不同？
             * 这里会把不属于原生变量的对象给删除掉（这里被删除掉的可能是父线程继承下来的，也可能是异步任务在执行时产生的新值）
             */
            if (!backupMap.containsKey(threadLocal)) {
                iterator.remove();
                threadLocal.superRemove();
            }
        }

        // 同样调用这个方法，进行值的恢复
        setTtlValuesTo(backupMap);
    }

    // 真正给当前子线程赋值的方法，对应上面的setTtlValuesTo方法
    private static void setTtlValuesTo(@Nonnull Map<TransmittableThreadLocal<?>, Object> ttlValues) {
        for (Map.Entry<TransmittableThreadLocal<?>, Object> entry : ttlValues.entrySet()) {
            @SuppressWarnings("unchecked")
            TransmittableThreadLocal<Object> threadLocal = (TransmittableThreadLocal<Object>) entry.getKey();
            threadLocal.set(entry.getValue()); //赋值，注意，从这里开始，子线程的holder里的值会被重新赋值刷新，可以参照上面ttl的set方法的实现
        }
    }

ok，到这里基本上把 TTL 比较核心的代码看完了，下面整理下整个流程，这是官方给出的时序图：

img

4.1、@Async 注解

这个类的 value 参数简直太友好了：

img

五处调用的地方，其中四处都是注释。

有效的调用就这一个地方，直接先打上断点再说：

org.springframework.scheduling.annotation.AnnotationAsyncExecutionInterceptor#getExecutorQualifier

发起调用之后，果然跑到了断点这个地方：

img

顺着断点往下调试，就会来到这个地方：

org.springframework.aop.interceptor.AsyncExecutionAspectSupport#determineAsyncExecutor

img

这个代码结构非常的清晰。

编号为 ① 的地方，是获取对应方法上的 @Async 注解的 value 值。这个值其实就是 bean 名称，如果不为空则从 Spring 容器中获取对应的 bean。

如果 value 是没有值的，也就是我们 Demo 的这种情况，会走到编号为 ② 的地方。

这个地方就是我要找的默认的线程池。

最后，不论是默认的线程池还是 Spring 容器中我们自定义的线程池。

都会以方法为维度，在 map 中维护方法和线程池的映射关系。

也就是编号为 ③ 的这一步，代码中的 executors 就是一个 map：

img

所以，我要找的东西，就是编号为 ② 的这个地方的逻辑。

这里面主要是一个 defaultExecutor 对象：

img

img

最终你会调试到这个地方来：

org.springframework.aop.interceptor.AsyncExecutionAspectSupport#getDefaultExecutor

img

这个代码就有点意思了，就是从 BeanFactory 里面获取一个默认的线程池相关的 Bean 出来。流程很简单，日志也打印的很清楚，就不赘述了。

5、修改方案

添加异步线程池, 使用 TtlExecutors.getTtlExecutor 修饰, 其余不变

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Component
public class AsyncThreadPoolConfiguration implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
        threadPool.setCorePoolSize(2);
        threadPool.setMaxPoolSize(5);
        threadPool.setQueueCapacity(20);
        threadPool.setKeepAliveSeconds(60);
        threadPool.setThreadNamePrefix("AsyncThread: ");
        threadPool.initialize();
        return TtlExecutors.getTtlExecutor(threadPool);
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}

img

本地验证, 虽然没有指定 value 但是还是进到了 配置的异步线程池中 会被 TtlExecutors 方法修饰, 获取上线文信息应该不会再有问题, 上线添加相关日志

6、总结

  1. 上线接口, 需要添加关键性日志, 以方便排查问题, 但是要主要日志数量控制
  2. 使用 @Asnyc 注解前需要关注到其余组件的配合使用
  3. 使用 ttl 获取上下文信息异常时, 不能简单传参解决问题, 以免后续 dubbo 等其他调用是其他字段信息异常

参考文档

TransmittableThreadLocal github : 传送门

ThreadLocal 系列（三）-TransmittableThreadLocal 的使用及原理解析 : 传送门

别问了，我真的不喜欢这个注解 : 传送门