ThreadLocal跨线程问题

港控/mmm° 2023-07-17 06:56 1阅读 0赞

## 1、问题 ##

通常复杂的处理流程中，我们会使用一些异步处理的手段，那么这种场景下ThreadLocal即可能出现获取失败的问题。

public class ThreadLocalTest { 
    
        public static void main(String[] args) { 
            ThreadLocal<String> threadLocal = new ThreadLocal<>();
            threadLocal.set("A");
            System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
            new Thread(()->{ 
                System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
            }).start();
        }
    }

![在这里插入图片描述][20200320150808524.png]

## 2、InheritableThreadLocal ##

直接使用ThreadLocal，在跨线程时时无法获取到ThreadLocal的。  
在lang包还有一个继承自ThreadLocal的类InheritableThreadLocal

public class InheritableThreadLocal<T> extends ThreadLocal<T> { 
        /** * Computes the child's initial value for this inheritable thread-local * variable as a function of the parent's value at the time the child * thread is created. This method is called from within the parent * thread before the child is started. * <p> * This method merely returns its input argument, and should be overridden * if a different behavior is desired. * * @param parentValue the parent thread's value * @return the child thread's initial value */
        protected T childValue(T parentValue) { 
            return parentValue;
        }
    
        /** * Get the map associated with a ThreadLocal. * * @param t the current thread */
        ThreadLocalMap getMap(Thread t) { 
           return t.inheritableThreadLocals;
        }
    
        /** * Create the map associated with a ThreadLocal. * * @param t the current thread * @param firstValue value for the initial entry of the table. */
        void createMap(Thread t, T firstValue) { 
            t.inheritableThreadLocals = new ThreadLocalMap(this, firstValue);
        }
    }

重写了getMap和createMap方法，让每次get、set操作都是对inheritableThreadLocals进行操作。  
在Thread的init方法中有个判断，若是父线程的inheritableThreadLocals不为空，则将其复制到子线程。  
![在这里插入图片描述][20200320151301608.png]  
也就是说我们使用InheritableThreadLocal，只要新建线程就可以让ThreadLocal在子父线程之间传递。

public class ThreadLocalTest { 
    
        public static void main(String[] args) { 
            ThreadLocal<String> threadLocal = new InheritableThreadLocal<>();
            threadLocal.set("A");
            System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
            new Thread(()->{ 
                System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
    
                new Thread(()->{ 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }).start();
            }).start();
        }
    }

![在这里插入图片描述][20200320151501781.png]

## 3、线程池复用Thread导致的问题 ##

并不代表InheritableThreadLocal就可以保证子父线程之间正确传递ThreadLocal对象。inheritableThreadLocals的复制操作只有在新创建Thread对象的时候才会触发。而我们通常不会在项目中new Thread，而是使用线程池，线程池的Thread对象是复用的。  
将上面的例子改成使用线程池  
使用固定大小的线程池，当固定大小为2时没问题，两次使用的不是同一个Thread，每次都成功复制了正确的ThreadLocal。

public class ThreadLocalTest1 { 
    
        public static void main(String[] args) throws InterruptedException { 
            Executor threadPool = Executors.newFixedThreadPool(2);
    
            ThreadLocal<String> threadLocal = new InheritableThreadLocal<>();
            threadLocal.set("A");
            System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
    
    
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
    
            threadLocal.set("B");
    
            Thread.sleep(200);
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
        }
    }

![在这里插入图片描述][2020032015212064.png]  
而将线程池固定大小设为1时会发现，使用同一个Thread对象，第二次并不会触发重新复制ThreadLocal对象，还是以前的A。

public class ThreadLocalTest1 { 
    
        public static void main(String[] args) throws InterruptedException { 
            Executor threadPool = Executors.newFixedThreadPool(1);
    
            ThreadLocal<String> threadLocal = new InheritableThreadLocal<>();
            threadLocal.set("A");
            System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
    
    
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
    
            threadLocal.set("B");
    
            Thread.sleep(200);
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
        }
    }

![在这里插入图片描述][2020032015221647.png]

## 4、解决办法 ##

**阿里的TransmittableThreadLocal**  
引入transmittable-thread-local依赖

<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>transmittable-thread-local</artifactId>
    			<version>2.11.4</version>
    		</dependency>

使用TransmittableThreadLocal继续改造上面的例子

public class ThreadLocalTest2 { 
    
        public static void main(String[] args) throws InterruptedException { 
            Executor threadPool = TtlExecutors.getTtlExecutor(Executors.newFixedThreadPool(1));
    
            ThreadLocal<String> threadLocal = new TransmittableThreadLocal<>();
            threadLocal.set("A");
            System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
    
    
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
    
            threadLocal.set("B");
    
            Thread.sleep(200);
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+"-getThreadLocal:"+threadLocal.get());
                }
            });
        }
    }

![在这里插入图片描述][20200320152939329.png]  
两个关键点：  
（1）需要使用transmittable-thread-local包中的方法包装线程池，否则即使使用TransmittableThreadLocal也不会起作用  
（2）使用TransmittableThreadLocal对象来保存线程变量。  
简单分析一下这个东东：  
TransmittableThreadLocal继承自InheritableThreadLocal  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjE0MjA0Mg_size_16_color_FFFFFF_t_70]  
TransmittableThreadLocal中有个holder的静态对象，对ThreadLocal进行get、set、remove等操作时其实是在对这个holder进行操作。用到了WeakHashMap，这个其实和JDK1.8以前的老版hashmap很相似，不同的是Entry对象是使用WeakReference包装的，这应该个原生ThreadLocal中的ThreadLocalMap使用WeakReference是一样的道理。（[ThreadLocal为什么使用WeakReference][ThreadLocal_WeakReference]）

private static InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>> holder = new InheritableThreadLocal<WeakHashMap<TransmittableThreadLocal<Object>, ?>>() { 
            protected WeakHashMap<TransmittableThreadLocal<Object>, ?> initialValue() { 
                return new WeakHashMap();
            }
    
            protected WeakHashMap<TransmittableThreadLocal<Object>, ?> childValue(WeakHashMap<TransmittableThreadLocal<Object>, ?> parentValue) { 
                return new WeakHashMap(parentValue);
            }
        };

然后看包装线程池是在干什么

@Nullable
        public static Executor getTtlExecutor(@Nullable Executor executor) { 
            return (Executor)(!TtlAgent.isTtlAgentLoaded() && null != executor && !(executor instanceof TtlEnhanced)?new ExecutorTtlWrapper(executor):executor);
        }

返回了一个ExecutorTtlWrapper对象，在调用execute时将Runnable对象换成了TtlRunnable。

class ExecutorTtlWrapper implements Executor, TtlWrapper<Executor>, TtlEnhanced { 
        private final Executor executor;
    
        ExecutorTtlWrapper(@NonNull Executor executor) { 
            this.executor = executor;
        }
    
        public void execute(@NonNull Runnable command) { 
            this.executor.execute(TtlRunnable.get(command));
        }
    
        @NonNull
        public Executor unwrap() { 
            return this.executor;
        }
    }

TtlRunnable中改造了原生Runnable的run方法

public final class TtlRunnable implements Runnable, TtlWrapper<Runnable>, TtlEnhanced, TtlAttachments { 
        private final AtomicReference<Object> capturedRef = new AtomicReference(Transmitter.capture());
        private final Runnable runnable;
        private final boolean releaseTtlValueReferenceAfterRun;
        private final TtlAttachmentsDelegate ttlAttachment = new TtlAttachmentsDelegate();
    
        private TtlRunnable(@NonNull Runnable runnable, boolean releaseTtlValueReferenceAfterRun) { 
            this.runnable = runnable;
            this.releaseTtlValueReferenceAfterRun = releaseTtlValueReferenceAfterRun;
        }
    
        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!");
            }
        }
        ......
     }

Transmitter.capture()其实就是在获取TransmittableThreadLocal中holder的副本，对该副本使用AtomicReference进行包装，方便保证原子性。  
调用run时，从capturedRef获取副本，使用cas将其更新为null，若cas失败则抛出异常IllegalStateException(“TTL value reference is released after run!”)，使用cas来保证每次取到的都是最新的副本。  
因此包装线程池的作用就是将以前每次新建Thread对象才拷贝inheritableThreadLocals的机制变成了每次新建Runnable的时候拷贝副本，从而保证线程池中子父线程之间ThreadLocal对象的传递。

这种在子父线程之间传递上下文的操作其实在很多框架中都有，如sleuth：  
以前一直很好奇，为什么异步调用时调用链路依然是完整的，很神奇。sleuth中使用一个叫ExecutorBeanPostProcessor的后置处理器包装了所有的线程池，例如将ThreadPoolTaskExecutor包装为LazyTraceThreadPoolTaskExecutor。LazyTraceThreadPoolTaskExecutor重写了线程池的execute、submit等方法，当我们传入参数Runnable或者Callable时会被包装为TraceRunnable或TraceCallable。

[20200320150808524.png]: /images/20230528/8286d143286d4ce3a84d6c64b7c300f8.png
[20200320151301608.png]: /images/20230528/d8a074d1cccc45ae9577d80afd77edc4.png
[20200320151501781.png]: /images/20230528/9f930c6ec11b428387b38b567f3c8afb.png
[2020032015212064.png]: /images/20230528/1d202bfc1e294907807c86fa2e1d6204.png
[2020032015221647.png]: /images/20230528/38ad9fee9aba4fa48002a89772cfbf55.png
[20200320152939329.png]: /images/20230528/0a6e5852fa3a47adb8b24c9fa99bf9cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjE0MjA0Mg_size_16_color_FFFFFF_t_70]: /images/20230528/25db98e4eab64fe796bf881cece5e1c3.png
[ThreadLocal_WeakReference]: https://blog.csdn.net/cyxinda/article/details/93709143