更新ThreadLocal分析文章

Author: coderbruis

JdkLearn/pom.xml

@@ -32,6 +32,13 @@
             <artifactId>netty-all</artifactId>
         </dependency>
 
+        <!-- 线程池上下文传递 -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>transmittable-thread-local</artifactId>
+            <version>2.14.5</version>
+        </dependency>
+
         <!-- redis依赖 -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

JdkLearn/src/main/java/com/learnjava/concurent/ThreadLocalTest.java

@@ -1,9 +1,25 @@
 package com.learnjava.concurent;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 public class ThreadLocalTest {
     public static void main(String[] args) throws InterruptedException {
+//        testThreadIsolation();
+//        testInitialValue();
+//        testSetGetRemove();
+//        testThreadLocalAsMapKey();
+//        testThreadPoolShouldRemove();
+        testHashIncrementDistribution();
+    }
+
+    /**
+     * 核心结果：同一个ThreadLocal对象，在不同线程中保存的是不同副本。
+     */
+    public static void testThreadIsolation() throws InterruptedException {
+        System.out.println("\n==== 1. thread isolation ====");
         int threads = 3;
         CountDownLatch countDownLatch = new CountDownLatch(threads);
 
@@ -20,7 +36,129 @@ public static void main(String[] args) throws InterruptedException {
             }, "thread - " + i).start();
         }
         countDownLatch.await();
+    }
+
+    /**
+     * initialValue只会在当前线程第一次get且没有值时触发。
+     */
+    public static void testInitialValue() throws InterruptedException {
+        System.out.println("\n==== 2. initialValue per thread ====");
+        ThreadLocal<StringBuilder> local = new ThreadLocal<StringBuilder>() {
+            @Override
+            protected StringBuilder initialValue() {
+                System.out.printf("%s init value\n", Thread.currentThread().getName());
+                return new StringBuilder(Thread.currentThread().getName());
+            }
+        };
+
+        Runnable task = () -> {
+            System.out.printf("%s first get: %s\n", Thread.currentThread().getName(), local.get());
+            System.out.printf("%s second get: %s\n", Thread.currentThread().getName(), local.get());
+        };
+
+        Thread threadA = new Thread(task, "thread-A");
+        Thread threadB = new Thread(task, "thread-B");
+        threadA.start();
+        threadB.start();
+        threadA.join();
+        threadB.join();
+    }
+
+    /**
+     * set/get/remove是ThreadLocal最常用的生命周期。
+     */
+    public static void testSetGetRemove() {
+        System.out.println("\n==== 3. set get remove ====");
+        ThreadLocal<String> local = new ThreadLocal<String>() {
+            @Override
+            protected String initialValue() {
+                return "init";
+            }
+        };
+
+        System.out.println("first get: " + local.get());
+        local.set("changed");
+        System.out.println("after set: " + local.get());
+        local.remove();
+        System.out.println("after remove, get again: " + local.get());
+    }
+
+    /**
+     * 原理要点：值不是存在ThreadLocal对象里，而是存在当前线程的ThreadLocalMap里。
+     * ThreadLocal实例本身作为key，所以同一个线程可以给不同ThreadLocal保存不同值。
+     */
+    public static void testThreadLocalAsMapKey() {
+        System.out.println("\n==== 4. ThreadLocal instance as key ====");
+        ThreadLocal<String> userLocal = new ThreadLocal<String>();
+        ThreadLocal<String> traceLocal = new ThreadLocal<String>();
+
+        userLocal.set("user-1001");
+        traceLocal.set("trace-abc");
+
+        System.out.printf("same thread:%s, userLocal:%s, traceLocal:%s\n",
+                Thread.currentThread().getName(),
+                userLocal.get(),
+                traceLocal.get());
 
+        userLocal.remove();
+        traceLocal.remove();
+    }
+
+    /**
+     * 线程池会复用线程；如果任务结束后不remove，后续任务可能读到上一个任务遗留的值。
+     */
+    public static void testThreadPoolShouldRemove() throws InterruptedException {
+        System.out.println("\n==== 5. thread pool should remove ====");
+        ThreadLocal<String> requestIdLocal = new ThreadLocal<String>();
+        ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+        executorService.execute(() -> {
+            requestIdLocal.set("request-1");
+            System.out.printf("%s set value: %s\n",
+                    Thread.currentThread().getName(),
+                    requestIdLocal.get());
+        });
+
+        executorService.execute(() -> {
+            System.out.printf("%s read old value: %s\n",
+                    Thread.currentThread().getName(),
+                    requestIdLocal.get());
+            requestIdLocal.remove();
+        });
+
+        executorService.execute(() -> System.out.printf("%s after remove: %s\n",
+                Thread.currentThread().getName(),
+                requestIdLocal.get()));
+
+        executorService.shutdown();
+        executorService.awaitTermination(3, TimeUnit.SECONDS);
+
+    }
+
+    /**
+     * 验证ThreadLocal中0x61c88647这个hash增量的分布效果。
+     * ThreadLocalMap长度是2的幂，定位下标时使用 hash & (len - 1)。
+     */
+    public static void testHashIncrementDistribution() {
+        System.out.println("\n==== 6. hash increment distribution ====");
+        int hashIncrement = 0x61c88647;
+
+        printHashIndexSequence(hashIncrement, 16);
+        printHashIndexSequence(hashIncrement, 32);
+    }
+
+    private static void printHashIndexSequence(int hashIncrement, int len) {
+        int hash = 0;
+
+        System.out.printf("len = %d, HASH_INCREMENT & (len - 1) = %d\n",
+                len,
+                hashIncrement & (len - 1));
+
+        for (int i = 0; i < len; i++) {
+            int index = hash & (len - 1);
+            System.out.printf("%2d -> hash: 0x%08x, index: %2d\n", i, hash, index);
+            hash += hashIncrement;
+        }
     }
 
     private static class InnerClass {
@@ -54,11 +192,6 @@ public void set(String word) {
     }
 
     private static class Counter {
-        private static ThreadLocal<StringBuilder> counter = new ThreadLocal<StringBuilder>() {
-            @Override
-            protected StringBuilder initialValue() {
-                return new StringBuilder();
-            }
-        };
+        private static ThreadLocal<StringBuilder> counter = ThreadLocal.withInitial(StringBuilder::new);
     }
 }

JdkLearn/src/main/java/com/learnjava/concurent/TransmittableThreadLocalDemo.java

@@ -0,0 +1,91 @@
+package com.learnjava.concurent;
+
+import com.alibaba.ttl.TransmittableThreadLocal;
+import com.alibaba.ttl.threadpool.TtlExecutors;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 演示 TransmittableThreadLocal 在线程池中的上下文传递。
+ */
+public class TransmittableThreadLocalDemo {
+
+    private static final ThreadLocal<String> NORMAL_THREAD_LOCAL = new ThreadLocal<>();
+    private static final InheritableThreadLocal<String> INHERITABLE_THREAD_LOCAL = new InheritableThreadLocal<>();
+    private static final TransmittableThreadLocal<String> TTL = new TransmittableThreadLocal<>();
+
+    public static void main(String[] args) throws InterruptedException {
+        testThreadLocalInThreadPool();
+        testInheritableThreadLocalInThreadPool();
+        testTransmittableThreadLocalInThreadPool();
+    }
+
+    /**
+     * 普通 ThreadLocal 不能跨线程传递值。
+     */
+    private static void testThreadLocalInThreadPool() throws InterruptedException {
+        System.out.println("\n==== 1. normal ThreadLocal ====");
+        ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+        NORMAL_THREAD_LOCAL.set("normal-context");
+        executorService.execute(() -> printValue("normal", NORMAL_THREAD_LOCAL.get()));
+
+        shutdown(executorService);
+        NORMAL_THREAD_LOCAL.remove();
+    }
+
+    /**
+     * InheritableThreadLocal 只在线程创建时传递，线程池复用线程时容易读到旧值。
+     */
+    private static void testInheritableThreadLocalInThreadPool() throws InterruptedException {
+        System.out.println("\n==== 2. InheritableThreadLocal ====");
+        ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+        INHERITABLE_THREAD_LOCAL.set("parent-context-1");
+        executorService.execute(() -> printValue("first task", INHERITABLE_THREAD_LOCAL.get()));
+        sleepQuietly(200);
+
+        INHERITABLE_THREAD_LOCAL.set("parent-context-2");
+        executorService.execute(() -> printValue("second task", INHERITABLE_THREAD_LOCAL.get()));
+
+        shutdown(executorService);
+        INHERITABLE_THREAD_LOCAL.remove();
+    }
+
+    /**
+     * TransmittableThreadLocal 通过包装线程池，在任务提交时捕获上下文，在执行时恢复上下文。
+     */
+    private static void testTransmittableThreadLocalInThreadPool() throws InterruptedException {
+        System.out.println("\n==== 3. TransmittableThreadLocal ====");
+        ExecutorService rawExecutorService = Executors.newFixedThreadPool(1);
+        ExecutorService ttlExecutorService = TtlExecutors.getTtlExecutorService(rawExecutorService);
+
+        TTL.set("ttl-context-1");
+        ttlExecutorService.execute(() -> printValue("first ttl task", TTL.get()));
+
+        TTL.set("ttl-context-2");
+        ttlExecutorService.execute(() -> printValue("second ttl task", TTL.get()));
+
+        shutdown(ttlExecutorService);
+        TTL.remove();
+    }
+
+    private static void printValue(String scene, String value) {
+        System.out.printf("%s, thread=%s, value=%s%n", scene, Thread.currentThread().getName(), value);
+    }
+
+    private static void shutdown(ExecutorService executorService) throws InterruptedException {
+        executorService.shutdown();
+        executorService.awaitTermination(3, TimeUnit.SECONDS);
+    }
+
+    private static void sleepQuietly(long millis) {
+        try {
+            TimeUnit.MILLISECONDS.sleep(millis);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}