InheritableThreadLocal 和线程池

Posted 2023-03-15

【中文标题】InheritableThreadLocal 和线程池

【英文标题】：InheritableThreadLocal and thread pools

【发布时间】：2011-11-09 22:19:45

【问题描述】：

我有一个我认为没有解决方案的问题，但无论如何我都会在这里尝试。我的应用程序使用线程池，并且该池中的一些线程具有可继承的线程局部变量。我已经扩展了 ThreadPoolExecutor 类，以在线程完成执行时从本质上清除线程局部变量（在 afterExecute 回调方法中）。

我知道当你有一个 InheritableThreadLocal 变量时，在初始化线程时调用 childValue() 方法以从父线程获取 ThreadLocal 变量的值。但是，在我的情况下，下次使用线程时（在使用一次之后），InheritableThreadLocal 变量的值为 null（因为它之前在 afterExecute 中被清除）。有没有办法在 beforeExecute 中访问父线程的线程局部变量，这样我就可以在本质上模拟 InheritableThreadLocal 中的 childValue 方法在创建线程时所做的事情。

【问题讨论】：

如果以后需要，为什么要清除子线程的线程本地值？下一次运行时需要的值是否来自不同的父线程，或者自第一次创建子线程以来它们在父线程中是否发生了变化？

@Bert - 这正是原因。基本上我将请求 id 存储在线程局部变量中，该变量会在子线程的多次使用中发生变化。

Spring security 为类似的问题做了一些解决方案，但我还没弄清楚到底是什么！

【参考方案1】：

我遇到了类似的问题，最终创建了一个 ForkListeningExecutorService，它包装了一个 ExecutorService（参见 here）。每当任务提交给执行程序或结束时，包装器都会向侦听器发送事件。然后，您可以使用侦听器在线程之间传递您想要的任何内容。这是 Listener 的样子：

  /**
   * Listener that will receive the events around task submission.
   *
   */
  public interface ExecutorServiceListener 

    /**
     * Will be called <b>before</b> any task is submitted to the service. This method will therefore run on the original "parent" thread.
     */
    default void beforeTaskSubmission() 
      // to be overridden by implementations
    

    /**
     * Will be called <b>after</b> a task is submitted to the service and <b>before</b> the actual execution starts. This method will therefore run on the new "child" thread.
     */
    default void afterTaskSubmission() 
      // to be overridden by implementations
    

    /**
     * Will be called <b>before</b> a submitted task ends no matter if an exception was thrown or not. This method will therefore run on the new "child" thread just before it's
     * released.
     */
    default void beforeTaskEnds() 
      // to be overridden by implementations
    
  

因此，您可以简单地包装它（或任何其他 ExecutorService 实现），而不是扩展 ThreadPoolExecutor，并像这样在侦听器中传递状态：

public class ForkListeningExecutorServiceExample 

  private static ThreadLocal<String> threadLocal = new ThreadLocal<>();

  private static void printThreadMessage(final String message) 
    System.out.println(message
        + ", current thread: " + Thread.currentThread().getName()
        + ", value from threadLocal: " + threadLocal.get());
  

  public static void main(final String[] args) throws Exception 

    threadLocal.set("MY_STATE");

    final ExecutorService executorService = new ForkListeningExecutorService(
        Executors.newCachedThreadPool(),
        new ExecutorServiceListener() 

          private String valueToShare;

          @Override
          public void beforeTaskSubmission() 
            valueToShare = threadLocal.get();
            printThreadMessage("The task is about to be submitted");
          

          @Override
          public void afterTaskSubmission() 
            threadLocal.set(valueToShare);
            printThreadMessage("The task has been submitted and will start now");
          

          @Override
          public void beforeTaskEnds() 
            threadLocal.set(null);
            printThreadMessage("The task has finished and thread will be released now");
          
        );

    executorService.submit(() -> 
      printThreadMessage("The task is running now");
    ).get();

    printThreadMessage("We are back on the main thread");
  

输出如下所示：

The task is about to be submitted, current thread: main, value from threadLocal: MY_STATE
The task has been submitted and will start now, current thread: pool-1-thread-1, value from threadLocal: MY_STATE
The task is running now, current thread: pool-1-thread-1, value from threadLocal: MY_STATE
The task has finished and thread will be released now, current thread: pool-1-thread-1, value from threadLocal: null
We are back on the main thread, current thread: main, value from threadLocal: MY_STATE

【参考方案2】：

runnable 的构造函数在调用者的线程中运行，而 run 方法在子线程中运行。您可以使用此事实将信息从父线程传输到子线程。见：

public class ThreadlocalApplication 
static public ThreadLocal<String> tenantId = new ThreadLocal<>();

public static void main(String[] args) throws ExecutionException, InterruptedException 
    ExecutorService executor = Executors.newCachedThreadPool();
    System.out.println(Thread.currentThread().getName());
    ThreadlocalApplication.tenantId.set("4711");
    executor.submit(new AsyncTask()).get();
    executor.shutdown();


static class AsyncTask implements Runnable 
    private String _tenantId;

    public AsyncTask() 
        System.out.println(Thread.currentThread().getName());
        _tenantId = ThreadlocalApplication.tenantId.get();
    

    @Override
    public void run() 
        ThreadlocalApplication.tenantId.set(_tenantId);
        System.out.println(Thread.currentThread().getName());
        System.out.println(ThreadlocalApplication.tenantId.get());
    

这就是结果

main
main
pool-1-thread-1
4711

【讨论】：

我非常喜欢这个答案，并认为它比@Stephen C 的答案更清晰、更直接。但是这里的私有变量（_tenantId）的线程安全呢？

【参考方案3】：

听起来这对于线程局部变量的“可继承”风格来说是一个糟糕的用例。

我的建议是只使用常规的TheadLocal 并明确地进行初始化；例如将父线程的初始值作为参数或其他东西传递给子线程。

（我建议您通过让子线程在启动时立即获取值来强制初始化子线程的本地线程。但这会冒竞争条件的风险；例如，如果父线程返回到池中在子线程开始执行之前。）

---

我想我要问的是是否有办法从子线程访问父线程的线程局部变量的值。

没有办法做到这一点。

而且，从你的其他 cmets 来看，我怀疑你的意思是正常意义上的“父”和“子”......父线程创建子线程。

但这里有一个想法。与其尝试在线程之间共享变量，不如共享一个固定值（例如请求 ID），并将其用作共享 Map 的键。使用Map 条目作为共享变量。

【讨论】：

感谢您的回复。我很好奇如何强制初始化子线程的线程本地。我想我要问的是是否有办法从子线程访问父线程的线程局部变量的值。

不幸的是，这个答案让问题悬而未决。如何将本地线程从一个线程转移到另一个线程。如果一个是另一个的孩子，那么 InheritableThreadLocals 可以很好地完成这项工作，但是如果是线程池，你会怎么做呢？

你为什么不赞成这个答案？对一个指出不存在好的解决方案的答案投反对票是一种粗鲁的做法……只是因为你不喜欢这个事实。当然，这不鼓励我帮助你！

这个问题必须有一个“真实”的答案。我知道以某种方式可以在线程之间共享相关器以记录由同一请求产生的事件。回答-1并不是粗鲁，我只是表示，根据我的说法，这不是答案。

注意到....我仍然不鼓励。这就是答案。只是不是你想要的答案。