java - 将 ThreadLocal 与 CompletableFuture 一起使用安全吗？

Question

ThreadLocal 将数据绑定(bind)到特定线程。对于 CompletableFuture，它使用线程池中的线程执行，该线程可能是不同的线程。

这是否意味着当CompletableFuture执行时，它可能无法从ThreadLocal获取数据？

Answer 1

each thread that accesses ThreadLocal (via its get or set method) has its own, independently initialized copy of the variable

所以不同的线程在使用ThreadLocal.get时会收到不同的值；另外，不同的线程在使用 ThreadLocal.set 时会设置自己的值； 不同线程之间 ThreadLocal 的内部/存储/自己的值不会重叠。

但是因为问题是关于与线程池结合使用的安全性，我将指出特定于该特殊组合的特定风险:

对于足够数量的调用，池中的线程有可能被重用(这就是池的全部意义:))。假设我们有 pool-thread1，它执行了任务 1，现在正在执行任务 2；如果任务 1 在完成其工作之前没有将其从 ThreadLocal 中删除，那么任务 2 将重用与任务 1 相同的 ThreadLocal 值!并且重用可能不是您想要的。

检查以下测试；他们可能会更好地证明我的观点。

package ro.go.adrhc.concurrent;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;

import static org.junit.Assert.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;

@Slf4j
class ThreadLocalTest {
    /**
     * Have 1 thread in order to have the 100% chance of 2 tasks using same copy of the ThreadLocal variable.
     */
    private ExecutorService es = Executors.newSingleThreadExecutor();
    private ThreadLocal<Double> cache = new ThreadLocal<>();
    /**
     * Random initialization isn't an alternative for proper cleaning!
     */
    private ThreadLocal<Double> cacheWithInitVal = ThreadLocal.withInitial(
            ThreadLocalRandom.current()::nextDouble);

    @Test
    void reuseThreadWithCleanup() throws ExecutionException, InterruptedException {
        var future1 = es.submit(() -> this.doSomethingWithCleanup(cache));
        var future2 = es.submit(() -> this.doSomethingWithCleanup(cache));
        assertNotEquals(future1.get(), future2.get()); // different runnable just used a different ThreadLocal value
    }

    @Test
    void reuseThreadWithoutInitVal() throws ExecutionException, InterruptedException {
        var future1 = es.submit(() -> this.doSomething(cache));
        var future2 = es.submit(() -> this.doSomething(cache));
        assertEquals(future1.get(), future2.get()); // different runnable just used the same ThreadLocal value
    }

    @Test
    void reuseThreadWithInitVal() throws ExecutionException, InterruptedException {
        var future1 = es.submit(() -> this.doSomething(cacheWithInitVal));
        var future2 = es.submit(() -> this.doSomething(cacheWithInitVal));
        assertEquals(future1.get(), future2.get()); // different runnable just used the same ThreadLocal value
    }

    private Double doSomething(ThreadLocal<Double> cache) {
        if (cache.get() == null) {
            // reusing ThreadLocal's value when not null
            cache.set(ThreadLocalRandom.current().nextDouble());
        }
        log.debug("thread: {}, cache: {}", Thread.currentThread().toString(), cache.get());
        return cache.get();
    }

    private Double doSomethingWithCleanup(ThreadLocal<Double> cache) {
        try {
            return doSomething(cache);
        } finally {
            cache.remove();
        }
    }
}