我看JAVA 之 Thread & ThreadLocal

注：基于jdk11

Thread

Thread是程序里执行的一个线程。JVM允许一个应用程序中可以有多个线程并发执行。
每一个线程都有一个优先级，高优先级的线程优于低优先级的线程先执行。同时，线程还可以被标记为守护线程。线程在被创建的时候优先级默认等同于创建者的优先级。

创建一个Thread通常有如下几种方式：

1. 继承Thread类，重写run()方法

   public class ConcreteThread extends Thread(){ public void run() { .... } } new ConcreteThread().start()

2. 实现Runnable接口，重写run()方法

   public class ConcreteThread implements Runnable(){ public void run() { .... } } new ConcreteThread().start()

3. 匿名类方式

   new Thread(new Runnable() { public void run() { .... } }).start()

实现了如下接口

1. Runnable 被FunctionalInterface注解的接口，定义了public abstract void run()方法供子类去实现。

几个重要的成员变量

1. private volatile String name; 被volatile修饰的name，每个线程必须有一个唯一的名字，方便调试，一般为 Thread-nextThreadNum()
2. private boolean daemon = false; 是否守护进程，默认为否
3. private boolean stillborn = false;
4. private long eetop;
5. private Runnable target; 执行目标
6. private ThreadGroup group; 线程组，默认为ecurity.getThreadGroup() 或 父线程所在组
7. private ClassLoader contextClassLoader;
8. private AccessControlContext inheritedAccessControlContext;
9. private static int threadInitNumber; 与Thread-拼接构成线程默认名称，private static synchronized int nextThreadNum()对其递增threadInitNumber++
10. ThreadLocal.ThreadLocalMap threadLocals = null;
11. ThreadLocal.ThreadLocalMap inheritableThreadLocals = null;
12. private final long stackSize; 为当前线程申请的栈空间，默认为0，取决于vm设计实现，有些vm会直接忽略此配置
13. private long nativeParkEventPointer;
14. private final long tid; 当前线程ID
15. private static long threadSeqNumber; 线程id计数器，private static synchronized long nextThreadID()对其递增++threadSeqNumber
16. private volatile int threadStatus;
17. volatile Object parkBlocker;
18. private volatile Interruptible blocker;
19. private final Object blockerLock = new Object();
20. public static final int MIN_PRIORITY = 1; 线程可以设置的最小优先级
    public static final int NORM_PRIORITY = 5;线程默认优先级
    public static final int MAX_PRIORITY = 10;线程可以设置的最大优先级
    线程的优先级会对应到不同操作系统的优先级，JVM不一定设置的优先级进行线程调度
21. 异常处理相关
    //当前线程异常处理handler，由volatile修饰
    private volatile UncaughtExceptionHandler uncaughtExceptionHandler;
    //所有线程缺省异常处理handler，由static volatile修饰
    private static volatile UncaughtExceptionHandler defaultUncaughtExceptionHandler;

几个重要的方法

1. 启动线程，JVM会调用当前线程的run方法

    public synchronized void start() { /** * This method is not invoked for the main method thread or "system" * group threads created/set up by the VM. Any new functionality added * to this method in the future may have to also be added to the VM. * * A zero status value corresponds to state "NEW". */ if (threadStatus != 0) throw new IllegalThreadStateException(); /* Notify the group that this thread is about to be started * so that it can be added to the group's list of threads * and the group's unstarted count can be decremented. */ group.add(this); boolean started = false; try { start0(); started = true; } finally { try { if (!started) { group.threadStartFailed(this); } } catch (Throwable ignore) { /* do nothing. If start0 threw a Throwable then it will be passed up the call stack */ } } } private native void start0();

2. 停止线程，已过时

    @Deprecated(since="1.2") public final void stop() { SecurityManager security = System.getSecurityManager(); if (security != null) { checkAccess(); if (this != Thread.currentThread()) { security.checkPermission(SecurityConstants.STOP_THREAD_PERMISSION); } } // A zero status value corresponds to "NEW", it can't change to // not-NEW because we hold the lock. if (threadStatus != 0) { resume(); // Wake up thread if it was suspended; no-op otherwise } // The VM can handle all thread states stop0(new ThreadDeath()); } private native void stop0(Object o);

   stop线程容易导致出现如下两种情况：

   1. 立即停止run()方法中剩余工作（包括在catch或finally语句），并抛出ThreadDeath异常(通常情况下此异常不需要显示的捕获)，可能会导致一些清理性工作的得不到执行，如文件流，数据库连接等的关闭。
   2. 会立即释放该线程所持有的所有的锁，导致数据得不到同步的处理，出现数据不一致的问题

3. 中断

    public void interrupt() { if (this != Thread.currentThread()) { checkAccess(); // thread may be blocked in an I/O operation synchronized (blockerLock) { Interruptible b = blocker; if (b != null) { interrupt0(); // set interrupt status b.interrupt(this); return; } } } // set interrupt status interrupt0(); } private native void interrupt0(); public static boolean interrupted() { return currentThread().isInterrupted(true); } private native boolean isInterrupted(boolean ClearInterrupted);

4. join 插队并阻塞当前执行线程，使用 loop + wait 的方式实现

    public final void join() throws InterruptedException { join(0); } public final synchronized void join(long millis) throws InterruptedException { long base = System.currentTimeMillis(); long now = 0; if (millis < 0) { throw new IllegalArgumentException("timeout value is negative"); } if (millis == 0) { while (isAlive()) { wait(0); } } else { while (isAlive()) { long delay = millis - now; if (delay <= 0) { break; } wait(delay); now = System.currentTimeMillis() - base; } } } public final native boolean isAlive();

5. suspend 与 resume 要成对出现，如果A线程访问某个资源x时suspend()，那么没有任何线程可以访问资源x直到A线程被resume()

    @Deprecated(since="1.2") public final void suspend() { checkAccess(); suspend0(); } @Deprecated(since="1.2") public final void resume() { checkAccess(); resume0(); }

线程状态及状态转换

1. 状态定义

    public enum State { NEW, RUNNABLE, BLOCKED, WAITING, TIMED_WAITING, TERMINATED; }

2. 状态图
   

3. 例子:

   package chapter02; public class TestThread { public static void main(String [] args) throws InterruptedException { final Thread thread0 = new Thread(new Runnable() { @Override public void run() { System.out.println("进入run"); try { System.out.printf("enter run(), thread0' state: %s\n", Thread.currentThread().getState()); Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); System.out.println("异常处理"); System.out.printf("on catch interrupt, thread0 isInterrupted or not ? %s \n", Thread.currentThread().isInterrupted()); System.out.printf("on catch interrupt, thread0' state: %s\n", Thread.currentThread().getState()); return; } System.out.println("退出run"); } }); Thread thread1 = new Thread(new Runnable() { @Override public void run() { System.out.println("进入thread1's run"); try { Thread.sleep(1000); System.out.printf("before interrupt, thread0 isInterrupted or nott ? %s \n", thread0.isInterrupted()); System.out.printf("enter thread1's run(), thread0' state: %s\n", thread0.getState()); Thread.sleep(1000); thread0.interrupt(); System.out.printf("after interrupt, thread0 isInterrupted or not ? %s \n", thread0.isInterrupted()); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("退出thread1's run"); } }); System.out.printf("after new(), thread0' state: %s\n", thread0.getState()); thread0.start(); System.out.printf("after start(), thread0' state: %s\n", thread0.getState()); thread1.start(); thread0.join(); System.out.printf("after join(), thread0' state: %s\n", thread0.getState()); System.out.println("退出"); } }

   打印结果如下：

   after new(), thread0' state: NEW after start(), thread0' state: RUNNABLE 进入run enter run(), thread0' state: RUNNABLE 进入thread1's run before interrupt, thread0 isInterrupted or nott ? false enter thread1's run(), thread0' state: TIMED_WAITING after interrupt, thread0 isInterrupted or not ? false 退出thread1's run 异常处理 on catch interrupt, thread0 isInterrupted or not ? false on catch interrupt, thread0' state: RUNNABLE after join(), thread0' state: TERMINATED 退出

异常捕获

1. 说明:
   //当前线程异常处理handler，由volatile修饰
   private volatile UncaughtExceptionHandler uncaughtExceptionHandler;
   //所有线程缺省异常处理handler，由static volatile修饰
   private static volatile UncaughtExceptionHandler defaultUncaughtExceptionHandler;

2. 例子:

   package chapter02; public class TestThread { public static void main(String [] args) throws InterruptedException { //全局异常处理器 Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException(Thread t, Throwable e) { System.out.println("-" + Thread.currentThread().getName()); String threadName = t.getName(); System.out.printf("global exception handler >> : current thread's name is %s, ", threadName); System.out.printf("the error is %s \n",e.getLocalizedMessage()); } }); final Thread thread0 = new Thread(new Runnable() { @Override public void run() { System.out.println("进入thread0's run"); System.out.printf("enter run(), thread0' state: %s\n", Thread.currentThread().getState()); try { Thread.sleep(5000); } catch (InterruptedException e) { // e.printStackTrace(); // System.out.println("异常处理"); // System.out.printf("on catch interrupt, thread0 isInterrupted or not ? %s \n", Thread.currentThread().isInterrupted()); // System.out.printf("on catch interrupt, thread0' state: %s\n", Thread.currentThread().getState()); // // return; throw new RuntimeException(e); } System.out.println("退出thread0's run"); } }); //thread0异常处理器 thread0.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException(Thread t, Throwable e) { System.out.println("-" + Thread.currentThread().getName()); String threadName = t.getName(); System.out.printf("thread0 exception handler >> : current thread's name is %s, ", threadName); System.out.printf("the error is %s \n",e.getLocalizedMessage()); } }); Thread thread1 = new Thread(new Runnable() { @Override public void run() { System.out.println("进入thread1's run"); try { Thread.sleep(1000); System.out.printf("before interrupt, thread0 isInterrupted or nott ? %s \n", thread0.isInterrupted()); System.out.printf("enter thread1's run(), thread0' state: %s\n", thread0.getState()); Thread.sleep(1000); thread0.interrupt(); System.out.printf("after interrupt, thread0 isInterrupted or not ? %s \n", thread0.isInterrupted()); Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("退出thread1's run"); } }); System.out.printf("after new(), thread0' state: %s\n", thread0.getState()); thread0.start(); System.out.printf("after start(), thread0' state: %s\n", thread0.getState()); thread1.setDaemon(true); thread1.start(); thread0.join(); thread1.join(); System.out.printf("after join(), thread0' state: %s\n", thread0.getState()); System.out.println("退出"); } }

   打印结果如下：

   after new(), thread0' state: NEW after start(), thread0' state: RUNNABLE 进入thread0's run enter run(), thread0' state: RUNNABLE 进入thread1's run before interrupt, thread0 isInterrupted or nott ? false enter thread1's run(), thread0' state: TIMED_WAITING after interrupt, thread0 isInterrupted or not ? true -Thread-0 thread0 exception handler >> : current thread's name is Thread-0, the error is java.lang.InterruptedException: sleep interrupted 退出thread1's run after join(), thread0' state: TERMINATED 退出

ThreadLocal

1. 说明:
   jdk1.2开始，为解决多线程程序的并发问题提供了一种新的思路ThreadLocal。使用这个工具类可以很简洁地编写出优美的多线程
   程序，ThreadLocal并不是一个Thread，而是Thread的局部变量。

2. 源码:

   public class ThreadLocal<T> { private final int threadLocalHashCode = nextHashCode(); private static AtomicInteger nextHashCode = new AtomicInteger(); private static final int HASH_INCREMENT = 0x61c88647; private static int nextHashCode() { return nextHashCode.getAndAdd(HASH_INCREMENT); } protected T initialValue() { return null; } public static <S> ThreadLocal<S> withInitial(Supplier<? extends S> supplier) { return new SuppliedThreadLocal<>(supplier); } public ThreadLocal() { } public T get() { Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) { ThreadLocalMap.Entry e = map.getEntry(this); if (e != null) { @SuppressWarnings("unchecked") T result = (T)e.value; return result; } } return setInitialValue(); } boolean isPresent() { Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); return map != null && map.getEntry(this) != null; } private T setInitialValue() { T value = initialValue(); Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) { map.set(this, value); } else { createMap(t, value); } if (this instanceof TerminatingThreadLocal) { TerminatingThreadLocal.register((TerminatingThreadLocal<?>) this); } return value; } //设置线程本地值，如果已经存在覆盖，否则为当前线程创建新的ThreadLocalMap，赋值给当前线程的threadLocals局部变量 public void set(T value) { Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) { map.set(this, value); } else { createMap(t, value); } } //删除本地值，不调用此方法在线程销毁后jvm也会回收，调用此方法后，如果多次访问get()方法可能导致多次触发initialValue() public void remove() { ThreadLocalMap m = getMap(Thread.currentThread()); if (m != null) { m.remove(this); } } ThreadLocalMap getMap(Thread t) { return t.threadLocals; } void createMap(Thread t, T firstValue) { t.threadLocals = new ThreadLocalMap(this, firstValue); } static ThreadLocalMap createInheritedMap(ThreadLocalMap parentMap) { return new ThreadLocalMap(parentMap); } T childValue(T parentValue) { throw new UnsupportedOperationException(); } static final class SuppliedThreadLocal<T> extends ThreadLocal<T> { private final Supplier<? extends T> supplier; SuppliedThreadLocal(Supplier<? extends T> supplier) { this.supplier = Objects.requireNonNull(supplier); } @Override protected T initialValue() { return supplier.get(); } } static class ThreadLocalMap { /** * The entries in this hash map extend WeakReference, using * its main ref field as the key (which is always a * ThreadLocal object). Note that null keys (i.e. entry.get() * == null) mean that the key is no longer referenced, so the * entry can be expunged from table. Such entries are referred to * as "stale entries" in the code that follows. */ static class Entry extends WeakReference<ThreadLocal<?>> { /** The value associated with this ThreadLocal. */ Object value; Entry(ThreadLocal<?> k, Object v) { super(k); value = v; } } private static final int INITIAL_CAPACITY = 16; private Entry[] table; private int size = 0; private int threshold; // Default to 0 private void setThreshold(int len) { threshold = len * 2 / 3; } private static int nextIndex(int i, int len) { return ((i + 1 < len) ? i + 1 : 0); } private static int prevIndex(int i, int len) { return ((i - 1 >= 0) ? i - 1 : len - 1); } ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) { table = new Entry[INITIAL_CAPACITY]; int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1); table[i] = new Entry(firstKey, firstValue); size = 1; setThreshold(INITIAL_CAPACITY); } private ThreadLocalMap(ThreadLocalMap parentMap) { Entry[] parentTable = parentMap.table; int len = parentTable.length; setThreshold(len); table = new Entry[len]; for (Entry e : parentTable) { if (e != null) { @SuppressWarnings("unchecked") ThreadLocal<Object> key = (ThreadLocal<Object>) e.get(); if (key != null) { Object value = key.childValue(e.value); Entry c = new Entry(key, value); int h = key.threadLocalHashCode & (len - 1); while (table[h] != null) h = nextIndex(h, len); table[h] = c; size++; } } } } private Entry getEntry(ThreadLocal<?> key) { int i = key.threadLocalHashCode & (table.length - 1); Entry e = table[i]; if (e != null && e.get() == key) return e; else return getEntryAfterMiss(key, i, e); } private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) { Entry[] tab = table; int len = tab.length; while (e != null) { ThreadLocal<?> k = e.get(); if (k == key) return e; if (k == null) expungeStaleEntry(i); else i = nextIndex(i, len); e = tab[i]; } return null; } private void set(ThreadLocal<?> key, Object value) { // We don't use a fast path as with get() because it is at // least as common to use set() to create new entries as // it is to replace existing ones, in which case, a fast // path would fail more often than not. Entry[] tab = table; int len = tab.length; int i = key.threadLocalHashCode & (len-1); for (Entry e = tab[i]; e != null; e = tab[i = nextIndex(i, len)]) { ThreadLocal<?> k = e.get(); if (k == key) { e.value = value; return; } if (k == null) { replaceStaleEntry(key, value, i); return; } } tab[i] = new Entry(key, value); int sz = ++size; if (!cleanSomeSlots(i, sz) && sz >= threshold) rehash(); } private void remove(ThreadLocal<?> key) { Entry[] tab = table; int len = tab.length; int i = key.threadLocalHashCode & (len-1); for (Entry e = tab[i]; e != null; e = tab[i = nextIndex(i, len)]) { if (e.get() == key) { e.clear(); expungeStaleEntry(i); return; } } } private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) { Entry[] tab = table; int len = tab.length; Entry e; // Back up to check for prior stale entry in current run. // We clean out whole runs at a time to avoid continual // incremental rehashing due to garbage collector freeing // up refs in bunches (i.e., whenever the collector runs). int slotToExpunge = staleSlot; for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) if (e.get() == null) slotToExpunge = i; // Find either the key or trailing null slot of run, whichever // occurs first for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); // If we find key, then we need to swap it // with the stale entry to maintain hash table order. // The newly stale slot, or any other stale slot // encountered above it, can then be sent to expungeStaleEntry // to remove or rehash all of the other entries in run. if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; // Start expunge at preceding stale entry if it exists if (slotToExpunge == staleSlot) slotToExpunge = i; cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; } // If we didn't find stale entry on backward scan, the // first stale entry seen while scanning for key is the // first still present in the run. if (k == null && slotToExpunge == staleSlot) slotToExpunge = i; } // If key not found, put new entry in stale slot tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value); // If there are any other stale entries in run, expunge them if (slotToExpunge != staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); } /** * Expunge a stale entry by rehashing any possibly colliding entries * lying between staleSlot and the next null slot. This also expunges * any other stale entries encountered before the trailing null. See * Knuth, Section 6.4 * * @param staleSlot index of slot known to have null key * @return the index of the next null slot after staleSlot * (all between staleSlot and this slot will have been checked * for expunging). */ private int expungeStaleEntry(int staleSlot) { Entry[] tab = table; int len = tab.length; // expunge entry at staleSlot tab[staleSlot].value = null; tab[staleSlot] = null; size--; // Rehash until we encounter null Entry e; int i; for (i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; tab[i] = null; size--; } else { int h = k.threadLocalHashCode & (len - 1); if (h != i) { tab[i] = null; // Unlike Knuth 6.4 Algorithm R, we must scan until // null because multiple entries could have been stale. while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; } } } return i; } private boolean cleanSomeSlots(int i, int n) { boolean removed = false; Entry[] tab = table; int len = tab.length; do { i = nextIndex(i, len); Entry e = tab[i]; if (e != null && e.get() == null) { n = len; removed = true; i = expungeStaleEntry(i); } } while ( (n >>>= 1) != 0); return removed; } private void rehash() { expungeStaleEntries(); // Use lower threshold for doubling to avoid hysteresis if (size >= threshold - threshold / 4) resize(); } private void resize() { Entry[] oldTab = table; int oldLen = oldTab.length; int newLen = oldLen * 2; Entry[] newTab = new Entry[newLen]; int count = 0; for (Entry e : oldTab) { if (e != null) { ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; // Help the GC } else { int h = k.threadLocalHashCode & (newLen - 1); while (newTab[h] != null) h = nextIndex(h, newLen); newTab[h] = e; count++; } } } setThreshold(newLen); size = count; table = newTab; } private void expungeStaleEntries() { Entry[] tab = table; int len = tab.length; for (int j = 0; j < len; j++) { Entry e = tab[j]; if (e != null && e.get() == null) expungeStaleEntry(j); } } } }