长期的应试教育让我花了太久才明白：学习其实是一件只与自己相关的事。 现在的我，更愿意主动探索那些真正能激发好奇心、让生活变得更有趣的事物。 如果你有能够启发我、让我产生兴趣的想法、机会或故事，欢迎写信给我： （ljr0115@outlook.com） 我会优先回复那些 与我的时间安排契合、或与我当下关注方向相吻合 的邮件。
AI 摘要 文章系统剖析ThreadLocal：从弱引用key、ThreadLocalMap结构、黄金分割哈希、线性探测解决冲突，到set/get、探测式与启发式清理、扩容机制，再延伸到InheritableThreadLocal及分布式链路traceId实战，覆盖原理、源码与项目踩坑。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: ThreadLocal 详解category: Javatag:Java并发前言全文共 10000+字，31 张图，这篇文章同样耗费了不少的时间和精力才创作完成，原创不易，请大家点点关注+在看，感谢。对于ThreadLocal，大家的第一反应可能是很简单呀，线程的变量副本，每个线程隔离。那这里有几个问题大家可以思考一下：ThreadLocal的 key 是弱引用，那么在 ThreadLocal.get()的时候，发生GC之后，key 是否为null？ThreadLocal中ThreadLocalMap的数据结构？ThreadLocalMap的Hash 算法？ThreadLocalMap中Hash 冲突如何解决？ThreadLocalMap的扩容机制？ThreadLocalMap中过期 key 的清理机制？探测式清理和启发式清理流程？ThreadLocalMap.set()方法实现原理？ThreadLocalMap.get()方法实现原理？项目中ThreadLocal使用情况？遇到的坑？……上述的一些问题你是否都已经掌握的很清楚了呢？本文将围绕这些问题使用图文方式来剖析ThreadLocal的点点滴滴。目录注明： 本文源码基于JDK 1.8ThreadLocal代码演示我们先看下ThreadLocal使用示例：public class ThreadLocalTest { private List<String> messages = Lists.newArrayList(); public static final ThreadLocal<ThreadLocalTest> holder = ThreadLocal.withInitial(ThreadLocalTest::new); public static void add(String message) { holder.get().messages.add(message); } public static List<String> clear() { List<String> messages = holder.get().messages; holder.remove(); System.out.println("size: " + holder.get().messages.size()); return messages; } public static void main(String[] args) { ThreadLocalTest.add("一枝花算不算浪漫"); System.out.println(holder.get().messages); ThreadLocalTest.clear(); } }打印结果：[一枝花算不算浪漫] size: 0ThreadLocal对象可以提供线程局部变量，每个线程Thread拥有一份自己的副本变量，多个线程互不干扰。ThreadLocal的数据结构Thread类有一个类型为ThreadLocal.ThreadLocalMap的实例变量threadLocals，也就是说每个线程有一个自己的ThreadLocalMap。ThreadLocalMap有自己的独立实现，可以简单地将它的key视作ThreadLocal，value为代码中放入的值（实际上key并不是ThreadLocal本身，而是它的一个弱引用）。每个线程在往ThreadLocal里放值的时候，都会往自己的ThreadLocalMap里存，读也是以ThreadLocal作为引用，在自己的map里找对应的key，从而实现了线程隔离。ThreadLocalMap有点类似HashMap的结构，只是HashMap是由数组+链表实现的，而ThreadLocalMap中并没有链表结构。我们还要注意Entry， 它的key是ThreadLocal<?> k ，继承自WeakReference， 也就是我们常说的弱引用类型。GC 之后 key 是否为 null？回应开头的那个问题， ThreadLocal 的key是弱引用，那么在ThreadLocal.get()的时候，发生GC之后，key是否是null？为了搞清楚这个问题，我们需要搞清楚Java的四种引用类型：强引用：我们常常 new 出来的对象就是强引用类型，只要强引用存在，垃圾回收器将永远不会回收被引用的对象，哪怕内存不足的时候软引用：使用 SoftReference 修饰的对象被称为软引用，软引用指向的对象在内存要溢出的时候被回收弱引用：使用 WeakReference 修饰的对象被称为弱引用，只要发生垃圾回收，若这个对象只被弱引用指向，那么就会被回收虚引用：虚引用是最弱的引用，在 Java 中使用 PhantomReference 进行定义。虚引用中唯一的作用就是用队列接收对象即将死亡的通知接着再来看下代码，我们使用反射的方式来看看GC后ThreadLocal中的数据情况：(下面代码来源自：https://blog.csdn.net/thewindkee/article/details/103726942 本地运行演示 GC 回收场景)public class ThreadLocalDemo { public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException, InterruptedException { Thread t = new Thread(()->test("abc",false)); t.start(); t.join(); System.out.println("--gc后--"); Thread t2 = new Thread(() -> test("def", true)); t2.start(); t2.join(); } private static void test(String s,boolean isGC) { try { new ThreadLocal<>().set(s); if (isGC) { System.gc(); } Thread t = Thread.currentThread(); Class<? extends Thread> clz = t.getClass(); Field field = clz.getDeclaredField("threadLocals"); field.setAccessible(true); Object ThreadLocalMap = field.get(t); Class<?> tlmClass = ThreadLocalMap.getClass(); Field tableField = tlmClass.getDeclaredField("table"); tableField.setAccessible(true); Object[] arr = (Object[]) tableField.get(ThreadLocalMap); for (Object o : arr) { if (o != null) { Class<?> entryClass = o.getClass(); Field valueField = entryClass.getDeclaredField("value"); Field referenceField = entryClass.getSuperclass().getSuperclass().getDeclaredField("referent"); valueField.setAccessible(true); referenceField.setAccessible(true); System.out.println(String.format("弱引用key:%s,值:%s", referenceField.get(o), valueField.get(o))); } } } catch (Exception e) { e.printStackTrace(); } } }结果如下：弱引用key:java.lang.ThreadLocal@433619b6,值:abc 弱引用key:java.lang.ThreadLocal@418a15e3,值:java.lang.ref.SoftReference@bf97a12 --gc后-- 弱引用key:null,值:def如图所示，因为这里创建的ThreadLocal并没有指向任何值，也就是没有任何引用：new ThreadLocal<>().set(s);所以这里在GC之后，key就会被回收，我们看到上面debug中的referent=null, 如果改动一下代码：这个问题刚开始看，如果没有过多思考，弱引用，还有垃圾回收，那么肯定会觉得是null。其实是不对的，因为题目说的是在做 ThreadLocal.get() 操作，证明其实还是有强引用存在的，所以 key 并不为 null，如下图所示，ThreadLocal的强引用仍然是存在的。如果我们的强引用不存在的话，那么 key 就会被回收，也就是会出现我们 value 没被回收，key 被回收，导致 value 永远存在，出现内存泄漏。ThreadLocal.set()方法源码详解ThreadLocal中的set方法原理如上图所示，很简单，主要是判断ThreadLocalMap是否存在，然后使用ThreadLocal中的set方法进行数据处理。代码如下：public void set(T value) { Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) map.set(this, value); else createMap(t, value); } void createMap(Thread t, T firstValue) { t.threadLocals = new ThreadLocalMap(this, firstValue); }主要的核心逻辑还是在ThreadLocalMap中的，一步步往下看，后面还有更详细的剖析。ThreadLocalMap Hash 算法既然是Map结构，那么ThreadLocalMap当然也要实现自己的hash算法来解决散列表数组冲突问题。int i = key.threadLocalHashCode & (len-1);ThreadLocalMap中hash算法很简单，这里i就是当前 key 在散列表中对应的数组下标位置。这里最关键的就是threadLocalHashCode值的计算，ThreadLocal中有一个属性为HASH_INCREMENT = 0x61c88647public 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); } static class ThreadLocalMap { 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); } } }每当创建一个ThreadLocal对象，这个ThreadLocal.nextHashCode 这个值就会增长 0x61c88647 。这个值很特殊，它是斐波那契数 也叫 黄金分割数。hash增量为 这个数字，带来的好处就是 hash 分布非常均匀。我们自己可以尝试下：可以看到产生的哈希码分布很均匀，这里不去细纠斐波那契具体算法，感兴趣的可以自行查阅相关资料。ThreadLocalMap Hash 冲突注明： 下面所有示例图中，绿色块Entry代表正常数据，灰色块代表Entry的key值为null，已被垃圾回收。白色块表示Entry为null。虽然ThreadLocalMap中使用了黄金分割数来作为hash计算因子，大大减少了Hash冲突的概率，但是仍然会存在冲突。HashMap中解决冲突的方法是在数组上构造一个链表结构，冲突的数据挂载到链表上，如果链表长度超过一定数量则会转化成红黑树。而 ThreadLocalMap 中并没有链表结构，所以这里不能使用 HashMap 解决冲突的方式了。如上图所示，如果我们插入一个value=27的数据，通过 hash 计算后应该落入槽位 4 中，而槽位 4 已经有了 Entry 数据。此时就会线性向后查找，一直找到 Entry 为 null 的槽位才会停止查找，将当前元素放入此槽位中。当然迭代过程中还有其他的情况，比如遇到了 Entry 不为 null 且 key 值相等的情况，还有 Entry 中的 key 值为 null 的情况等等都会有不同的处理，后面会一一详细讲解。这里还画了一个Entry中的key为null的数据（Entry=2 的灰色块数据），因为key值是弱引用类型，所以会有这种数据存在。在set过程中，如果遇到了key过期的Entry数据，实际上是会进行一轮探测式清理操作的，具体操作方式后面会讲到。ThreadLocalMap.set()详解ThreadLocalMap.set()原理图解看完了ThreadLocal hash 算法后，我们再来看set是如何实现的。往ThreadLocalMap中set数据（新增或者更新数据）分为好几种情况，针对不同的情况我们画图来说明。第一种情况： 通过hash计算后的槽位对应的Entry数据为空：这里直接将数据放到该槽位即可。第二种情况： 槽位数据不为空，key值与当前ThreadLocal通过hash计算获取的key值一致：这里直接更新该槽位的数据。第三种情况： 槽位数据不为空，往后遍历过程中，在找到Entry为null的槽位之前，没有遇到key过期的Entry：遍历散列数组，线性往后查找，如果找到Entry为null的槽位，则将数据放入该槽位中，或者往后遍历过程中，遇到了key 值相等的数据，直接更新即可。第四种情况： 槽位数据不为空，往后遍历过程中，在找到Entry为null的槽位之前，遇到key过期的Entry，如下图，往后遍历过程中，遇到了index=7的槽位数据Entry的key=null：散列数组下标为 7 位置对应的Entry数据key为null，表明此数据key值已经被垃圾回收掉了，此时就会执行replaceStaleEntry()方法，该方法含义是替换过期数据的逻辑，以index=7位起点开始遍历，进行探测式数据清理工作。初始化探测式清理过期数据扫描的开始位置：slotToExpunge = staleSlot = 7以当前staleSlot开始 向前迭代查找，找其他过期的数据，然后更新过期数据起始扫描下标slotToExpunge。for循环迭代，直到碰到Entry为null结束。如果找到了过期的数据，继续向前迭代，直到遇到Entry=null的槽位才停止迭代，如下图所示，slotToExpunge 被更新为 0：以当前节点(index=7)向前迭代，检测是否有过期的Entry数据，如果有则更新slotToExpunge值。碰到null则结束探测。以上图为例slotToExpunge被更新为 0。上面向前迭代的操作是为了更新探测清理过期数据的起始下标slotToExpunge的值，这个值在后面会讲解，它是用来判断当前过期槽位staleSlot之前是否还有过期元素。接着开始以staleSlot位置(index=7)向后迭代，如果找到了相同 key 值的 Entry 数据：从当前节点staleSlot向后查找key值相等的Entry元素，找到后更新Entry的值并交换staleSlot元素的位置(staleSlot位置为过期元素)，更新Entry数据，然后开始进行过期Entry的清理工作，如下图所示：向后遍历过程中，如果没有找到相同 key 值的 Entry 数据：从当前节点staleSlot向后查找key值相等的Entry元素，直到Entry为null则停止寻找。通过上图可知，此时table中没有key值相同的Entry。创建新的Entry，替换table[stableSlot]位置：替换完成后也是进行过期元素清理工作，清理工作主要是有两个方法：expungeStaleEntry()和cleanSomeSlots()，具体细节后面会讲到，请继续往后看。ThreadLocalMap.set()源码详解上面已经用图的方式解析了set()实现的原理，其实已经很清晰了，我们接着再看下源码：java.lang.ThreadLocal.ThreadLocalMap.set():private void set(ThreadLocal<?> key, Object value) { 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(); }这里会通过key来计算在散列表中的对应位置，然后以当前key对应的桶的位置向后查找，找到可以使用的桶。Entry[] tab = table; int len = tab.length; int i = key.threadLocalHashCode & (len-1);什么情况下桶才是可以使用的呢？k = key 说明是替换操作，可以使用碰到一个过期的桶，执行替换逻辑，占用过期桶查找过程中，碰到桶中Entry=null的情况，直接使用接着就是执行for循环遍历，向后查找，我们先看下nextIndex()、prevIndex()方法实现：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); }接着看剩下for循环中的逻辑：遍历当前key值对应的桶中Entry数据为空，这说明散列数组这里没有数据冲突，跳出for循环，直接set数据到对应的桶中如果key值对应的桶中Entry数据不为空 2.1 如果k = key，说明当前set操作是一个替换操作，做替换逻辑，直接返回 2.2 如果key = null，说明当前桶位置的Entry是过期数据，执行replaceStaleEntry()方法(核心方法)，然后返回for循环执行完毕，继续往下执行说明向后迭代的过程中遇到了entry为null的情况 3.1 在Entry为null的桶中创建一个新的Entry对象 3.2 执行++size操作调用cleanSomeSlots()做一次启发式清理工作，清理散列数组中Entry的key过期的数据 4.1 如果清理工作完成后，未清理到任何数据，且size超过了阈值(数组长度的 2/3)，进行rehash()操作 4.2 rehash()中会先进行一轮探测式清理，清理过期key，清理完成后如果size >= threshold - threshold / 4，就会执行真正的扩容逻辑(扩容逻辑往后看)接着重点看下replaceStaleEntry()方法，replaceStaleEntry()方法提供替换过期数据的功能，我们可以对应上面第四种情况的原理图来再回顾下，具体代码如下：java.lang.ThreadLocal.ThreadLocalMap.replaceStaleEntry():private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) { Entry[] tab = table; int len = tab.length; Entry e; int slotToExpunge = staleSlot; for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) if (e.get() == null) slotToExpunge = i; for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; if (slotToExpunge == staleSlot) slotToExpunge = i; cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; } if (k == null && slotToExpunge == staleSlot) slotToExpunge = i; } tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value); if (slotToExpunge != staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); }slotToExpunge表示开始探测式清理过期数据的开始下标，默认从当前的staleSlot开始。以当前的staleSlot开始，向前迭代查找，找到没有过期的数据，for循环一直碰到Entry为null才会结束。如果向前找到了过期数据，更新探测清理过期数据的开始下标为 i，即slotToExpunge=ifor (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)){ if (e.get() == null){ slotToExpunge = i; } }接着开始从staleSlot向后查找，也是碰到Entry为null的桶结束。如果迭代过程中，碰到 k == key，这说明这里是替换逻辑，替换新数据并且交换当前staleSlot位置。如果slotToExpunge == staleSlot，这说明replaceStaleEntry()一开始向前查找过期数据时并未找到过期的Entry数据，接着向后查找过程中也未发现过期数据，修改开始探测式清理过期数据的下标为当前循环的 index，即slotToExpunge = i。最后调用cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);进行启发式过期数据清理。if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; if (slotToExpunge == staleSlot) slotToExpunge = i; cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; }cleanSomeSlots()和expungeStaleEntry()方法后面都会细讲，这两个是和清理相关的方法，一个是过期key相关Entry的启发式清理(Heuristically scan)，另一个是过期key相关Entry的探测式清理。如果 k != key则会接着往下走，k == null说明当前遍历的Entry是一个过期数据，slotToExpunge == staleSlot说明，一开始的向前查找数据并未找到过期的Entry。如果条件成立，则更新slotToExpunge 为当前位置，这个前提是前驱节点扫描时未发现过期数据。if (k == null && slotToExpunge == staleSlot) slotToExpunge = i;往后迭代的过程中如果没有找到k == key的数据，且碰到Entry为null的数据，则结束当前的迭代操作。此时说明这里是一个添加的逻辑，将新的数据添加到table[staleSlot] 对应的slot中。tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value);最后判断除了staleSlot以外，还发现了其他过期的slot数据，就要开启清理数据的逻辑：if (slotToExpunge != staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);ThreadLocalMap过期 key 的探测式清理流程上面我们有提及ThreadLocalMap的两种过期key数据清理方式：探测式清理和启发式清理。我们先讲下探测式清理，也就是expungeStaleEntry方法，遍历散列数组，从开始位置向后探测清理过期数据，将过期数据的Entry设置为null，沿途中碰到未过期的数据则将此数据rehash后重新在table数组中定位，如果定位的位置已经有了数据，则会将未过期的数据放到最靠近此位置的Entry=null的桶中，使rehash后的Entry数据距离正确的桶的位置更近一些。操作逻辑如下：如上图，set(27) 经过 hash 计算后应该落到index=4的桶中，由于index=4桶已经有了数据，所以往后迭代最终数据放入到index=7的桶中，放入后一段时间后index=5中的Entry数据key变为了null如果再有其他数据set到map中，就会触发探测式清理操作。如上图，执行探测式清理后，index=5的数据被清理掉，继续往后迭代，到index=7的元素时，经过rehash后发现该元素正确的index=4，而此位置已经有了数据，往后查找离index=4最近的Entry=null的节点(刚被探测式清理掉的数据：index=5)，找到后移动index= 7的数据到index=5中，此时桶的位置离正确的位置index=4更近了。经过一轮探测式清理后，key过期的数据会被清理掉，没过期的数据经过rehash重定位后所处的桶位置理论上更接近i= key.hashCode & (tab.len - 1)的位置。这种优化会提高整个散列表查询性能。接着看下expungeStaleEntry()具体流程，我们还是以先原理图后源码讲解的方式来一步步梳理：我们假设expungeStaleEntry(3) 来调用此方法，如上图所示，我们可以看到ThreadLocalMap中table的数据情况，接着执行清理操作：第一步是清空当前staleSlot位置的数据，index=3位置的Entry变成了null。然后接着往后探测：执行完第二步后，index=4 的元素挪到 index=3 的槽位中。继续往后迭代检查，碰到正常数据，计算该数据位置是否偏移，如果被偏移，则重新计算slot位置，目的是让正常数据尽可能存放在正确位置或离正确位置更近的位置在往后迭代的过程中碰到空的槽位，终止探测，这样一轮探测式清理工作就完成了，接着我们继续看看具体实现源代码：private int expungeStaleEntry(int staleSlot) { Entry[] tab = table; int len = tab.length; tab[staleSlot].value = null; tab[staleSlot] = null; size--; 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; while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; } } } return i; }这里我们还是以staleSlot=3 来做示例说明，首先是将tab[staleSlot]槽位的数据清空，然后设置size--接着以staleSlot位置往后迭代，如果遇到k==null的过期数据，也是清空该槽位数据，然后size--ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; tab[i] = null; size--; }如果key没有过期，重新计算当前key的下标位置是不是当前槽位下标位置，如果不是，那么说明产生了hash冲突，此时以新计算出来正确的槽位位置往后迭代，找到最近一个可以存放entry的位置。int h = k.threadLocalHashCode & (len - 1); if (h != i) { tab[i] = null; while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; }这里是处理正常的产生Hash冲突的数据，经过迭代后，有过Hash冲突数据的Entry位置会更靠近正确位置，这样的话，查询的时候 效率才会更高。ThreadLocalMap扩容机制在ThreadLocalMap.set()方法的最后，如果执行完启发式清理工作后，未清理到任何数据，且当前散列数组中Entry的数量已经达到了列表的扩容阈值(len*2/3)，就开始执行rehash()逻辑：if (!cleanSomeSlots(i, sz) && sz >= threshold) rehash();接着看下rehash()具体实现：private void rehash() { expungeStaleEntries(); if (size >= threshold - threshold / 4) resize(); } 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); } }这里首先是会进行探测式清理工作，从table的起始位置往后清理，上面有分析清理的详细流程。清理完成之后，table中可能有一些key为null的Entry数据被清理掉，所以此时通过判断size >= threshold - threshold / 4 也就是size >= threshold * 3/4 来决定是否扩容。我们还记得上面进行rehash()的阈值是size >= threshold，所以当面试官套路我们ThreadLocalMap扩容机制的时候 我们一定要说清楚这两个步骤：接着看看具体的resize()方法，为了方便演示，我们以oldTab.len=8来举例：扩容后的tab的大小为oldLen * 2，然后遍历老的散列表，重新计算hash位置，然后放到新的tab数组中，如果出现hash冲突则往后寻找最近的entry为null的槽位，遍历完成之后，oldTab中所有的entry数据都已经放入到新的tab中了。重新计算tab下次扩容的阈值，具体代码如下：private void resize() { Entry[] oldTab = table; int oldLen = oldTab.length; int newLen = oldLen * 2; Entry[] newTab = new Entry[newLen]; int count = 0; for (int j = 0; j < oldLen; ++j) { Entry e = oldTab[j]; if (e != null) { ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; } 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; }ThreadLocalMap.get()详解上面已经看完了set()方法的源码，其中包括set数据、清理数据、优化数据桶的位置等操作，接着看看get()操作的原理。ThreadLocalMap.get()图解第一种情况： 通过查找key值计算出散列表中slot位置，然后该slot位置中的Entry.key和查找的key一致，则直接返回：第二种情况： slot位置中的Entry.key和要查找的key不一致：我们以get(ThreadLocal1)为例，通过hash计算后，正确的slot位置应该是 4，而index=4的槽位已经有了数据，且key值不等于ThreadLocal1，所以需要继续往后迭代查找。迭代到index=5的数据时，此时Entry.key=null，触发一次探测式数据回收操作，执行expungeStaleEntry()方法，执行完后，index 5,8的数据都会被回收，而index 6,7的数据都会前移。index 6,7前移之后，继续从 index=5 往后迭代，于是就在 index=6 找到了key值相等的Entry数据，如下图所示：ThreadLocalMap.get()源码详解java.lang.ThreadLocal.ThreadLocalMap.getEntry():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; }ThreadLocalMap过期 key 的启发式清理流程上面多次提及到ThreadLocalMap过期 key 的两种清理方式：探测式清理(expungeStaleEntry())、启发式清理(cleanSomeSlots())探测式清理是以当前Entry 往后清理，遇到值为null则结束清理，属于线性探测清理。而启发式清理被作者定义为：Heuristically scan some cells looking for stale entries.具体代码如下：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; }InheritableThreadLocal我们使用ThreadLocal的时候，在异步场景下是无法给子线程共享父线程中创建的线程副本数据的。为了解决这个问题，JDK 中还有一个InheritableThreadLocal类，我们来看一个例子：public class InheritableThreadLocalDemo { public static void main(String[] args) { ThreadLocal<String> ThreadLocal = new ThreadLocal<>(); ThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>(); ThreadLocal.set("父类数据:threadLocal"); inheritableThreadLocal.set("父类数据:inheritableThreadLocal"); new Thread(new Runnable() { @Override public void run() { System.out.println("子线程获取父类ThreadLocal数据：" + ThreadLocal.get()); System.out.println("子线程获取父类inheritableThreadLocal数据：" + inheritableThreadLocal.get()); } }).start(); } }打印结果：子线程获取父类ThreadLocal数据：null 子线程获取父类inheritableThreadLocal数据：父类数据:inheritableThreadLocal实现原理是子线程是通过在父线程中通过调用new Thread()方法来创建子线程，Thread#init方法在Thread的构造方法中被调用。在init方法中拷贝父线程数据到子线程中：private void init(ThreadGroup g, Runnable target, String name, long stackSize, AccessControlContext acc, boolean inheritThreadLocals) { if (name == null) { throw new NullPointerException("name cannot be null"); } if (inheritThreadLocals && parent.inheritableThreadLocals != null) this.inheritableThreadLocals = ThreadLocal.createInheritedMap(parent.inheritableThreadLocals); this.stackSize = stackSize; tid = nextThreadID(); }但InheritableThreadLocal仍然有缺陷，一般我们做异步化处理都是使用的线程池，而InheritableThreadLocal是在new Thread中的init()方法给赋值的，而线程池是线程复用的逻辑，所以这里会存在问题。当然，有问题出现就会有解决问题的方案，阿里巴巴开源了一个TransmittableThreadLocal组件就可以解决这个问题，这里就不再延伸，感兴趣的可自行查阅资料。ThreadLocal项目中使用实战ThreadLocal使用场景我们现在项目中日志记录用的是ELK+Logstash，最后在Kibana中进行展示和检索。现在都是分布式系统统一对外提供服务，项目间调用的关系可以通过 traceId 来关联，但是不同项目之间如何传递 traceId 呢？这里我们使用 org.slf4j.MDC 来实现此功能，内部就是通过 ThreadLocal 来实现的，具体实现如下：当前端发送请求到服务 A时，服务 A会生成一个类似UUID的traceId字符串，将此字符串放入当前线程的ThreadLocal中，在调用服务 B的时候，将traceId写入到请求的Header中，服务 B在接收请求时会先判断请求的Header中是否有traceId，如果存在则写入自己线程的ThreadLocal中。图中的requestId即为我们各个系统链路关联的traceId，系统间互相调用，通过这个requestId即可找到对应链路，这里还有会有一些其他场景：针对于这些场景，我们都可以有相应的解决方案，如下所示Feign 远程调用解决方案服务发送请求：@Component @Slf4j public class FeignInvokeInterceptor implements RequestInterceptor { @Override public void apply(RequestTemplate template) { String requestId = MDC.get("requestId"); if (StringUtils.isNotBlank(requestId)) { template.header("requestId", requestId); } } }服务接收请求：@Slf4j @Component public class LogInterceptor extends HandlerInterceptorAdapter { @Override public void afterCompletion(HttpServletRequest arg0, HttpServletResponse arg1, Object arg2, Exception arg3) { MDC.remove("requestId"); } @Override public void postHandle(HttpServletRequest arg0, HttpServletResponse arg1, Object arg2, ModelAndView arg3) { } @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { String requestId = request.getHeader(BaseConstant.REQUEST_ID_KEY); if (StringUtils.isBlank(requestId)) { requestId = UUID.randomUUID().toString().replace("-", ""); } MDC.put("requestId", requestId); return true; } }线程池异步调用，requestId 传递因为MDC是基于ThreadLocal去实现的，异步过程中，子线程并没有办法获取到父线程ThreadLocal存储的数据，所以这里可以自定义线程池执行器，修改其中的run()方法：public class MyThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { @Override public void execute(Runnable runnable) { Map<String, String> context = MDC.getCopyOfContextMap(); super.execute(() -> run(runnable, context)); } @Override private void run(Runnable runnable, Map<String, String> context) { if (context != null) { MDC.setContextMap(context); } try { runnable.run(); } finally { MDC.remove(); } } }使用 MQ 发送消息给第三方系统在 MQ 发送的消息体中自定义属性requestId，接收方消费消息后，自己解析requestId使用即可。
AI 摘要 文章介绍死锁概念、模拟代码、排查工具后，给出核心避免策略：资源有序分配法，即所有线程按统一顺序获取锁，破坏环路等待条件即可杜绝死锁。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结死锁的概念在多线程编程中，我们为了防止多线程竞争共享资源而导致数据错乱，都会在操作共享资源之前加上互斥锁，只有成功获得到锁的线程，才能操作共享资源，获取不到锁的线程就只能等待，直到锁被释放。那么，当两个线程为了保护两个不同的共享资源而使用了两个互斥锁，那么这两个互斥锁应用不当的时候，可能会造成两个线程都在等待对方释放锁，在没有外力的作用下，这些线程会一直相互等待，就没办法继续运行，这种情况就是发生了死锁。举个例子，樱舞拿了小美房间的钥匙，而樱舞在自己的房间里，小美拿了樱舞房间的钥匙，而小美也在自己的房间里。如果樱舞要从自己的房间里出去，必须拿到小美手中的钥匙，但是小美要出去，又必须拿到樱舞手中的钥匙，这就形成了死锁。死锁只有同时满足以下四个条件才会发生：互斥条件；持有并等待条件；不可剥夺条件；环路等待条件；互斥条件互斥条件是指多个线程不能同时使用同一个资源。比如下图，如果线程 A 已经持有的资源，不能再同时被线程 B 持有，如果线程 B 请求获取线程 A 已经占用的资源，那线程 B 只能等待，直到线程 A 释放了资源。持有并等待条件持有并等待条件是指，当线程 A 已经持有了资源 1，又想申请资源 2，而资源 2 已经被线程 C 持有了，所以线程 A 就会处于等待状态，但是线程 A 在等待资源 2 的同时并不会释放自己已经持有的资源 1。 不可剥夺条件不可剥夺条件是指，当线程已经持有了资源，在自己使用完之前不能被其他线程获取，线程 B 如果也想使用此资源，则只能在线程 A 使用完并释放后才能获取。环路等待条件环路等待条件指的是，在死锁发生的时候，两个线程获取资源的顺序构成了环形链。比如，线程 A 已经持有资源 2，而想请求资源 1，线程 B 已经获取了资源 1，而想请求资源 2，这就形成资源请求等待的环形图。模拟死锁问题的产生Talk is cheap. Show me the code.下面，我们用代码来模拟死锁问题的产生。首先，我们先创建 2 个线程，分别为线程 A 和 线程 B，然后有两个互斥锁，分别是 mutex_A 和 mutex_B，代码如下：pthread_mutex_t mutex_A = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex_B = PTHREAD_MUTEX_INITIALIZER; int main() { pthread_t tidA, tidB; //创建两个线程 pthread_create(&tidA, NULL, threadA_proc, NULL); pthread_create(&tidB, NULL, threadB_proc, NULL); pthread_join(tidA, NULL); pthread_join(tidB, NULL); printf("exit\n"); return 0; }接下来，我们看下线程 A 函数做了什么。//线程函数 A void *threadA_proc(void *data) { printf("thread A waiting get ResourceA \n"); pthread_mutex_lock(&mutex_A); printf("thread A got ResourceA \n"); sleep(1); printf("thread A waiting get ResourceB \n"); pthread_mutex_lock(&mutex_B); printf("thread A got ResourceB \n"); pthread_mutex_unlock(&mutex_B); pthread_mutex_unlock(&mutex_A); return (void *)0; }可以看到，线程 A 函数的过程：先获取互斥锁 A，然后睡眠 1 秒；再获取互斥锁 B，然后释放互斥锁 B；最后释放互斥锁 A；//线程函数 B void *threadB_proc(void *data) { printf("thread B waiting get ResourceB \n"); pthread_mutex_lock(&mutex_B); printf("thread B got ResourceB \n"); sleep(1); printf("thread B waiting get ResourceA \n"); pthread_mutex_lock(&mutex_A); printf("thread B got ResourceA \n"); pthread_mutex_unlock(&mutex_A); pthread_mutex_unlock(&mutex_B); return (void *)0; }可以看到，线程 B 函数的过程：先获取互斥锁 B，然后睡眠 1 秒；再获取互斥锁 A，然后释放互斥锁 A；最后释放互斥锁 B；然后，我们运行这个程序，运行结果如下：thread B waiting get ResourceB thread B got ResourceB thread A waiting get ResourceA thread A got ResourceA thread B waiting get ResourceA thread A waiting get ResourceB // 阻塞中。。。可以看到线程 B 在等待互斥锁 A 的释放，线程 A 在等待互斥锁 B 的释放，双方都在等待对方资源的释放，很明显，产生了死锁问题。利用工具排查死锁问题如果你想排查你的 Java 程序是否死锁，则可以使用 jstack 工具，它是 jdk 自带的线程堆栈分析工具。由于樱舞的死锁代码例子是 C 写的，在 Linux 下，我们可以使用 pstack + gdb 工具来定位死锁问题。pstack 命令可以显示每个线程的栈跟踪信息（函数调用过程），它的使用方式也很简单，只需要 pstack <pid> 就可以了。那么，在定位死锁问题时，我们可以多次执行 pstack 命令查看线程的函数调用过程，多次对比结果，确认哪几个线程一直没有变化，且是因为在等待锁，那么大概率是由于死锁问题导致的。我用 pstack 输出了我前面模拟死锁问题的进程的所有线程的情况，我多次执行命令后，其结果都一样，如下：$ pstack 87746 Thread 3 (Thread 0x7f60a610a700 (LWP 87747)): #0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 #1 0x0000003720e093ca in _L_lock_829 () from /lib64/libpthread.so.0 #2 0x0000003720e09298 in pthread_mutex_lock () from /lib64/libpthread.so.0 #3 0x0000000000400725 in threadA_proc () #4 0x0000003720e07893 in start_thread () from /lib64/libpthread.so.0 #5 0x00000037206f4bfd in clone () from /lib64/libc.so.6 Thread 2 (Thread 0x7f60a5709700 (LWP 87748)): #0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 #1 0x0000003720e093ca in _L_lock_829 () from /lib64/libpthread.so.0 #2 0x0000003720e09298 in pthread_mutex_lock () from /lib64/libpthread.so.0 #3 0x0000000000400792 in threadB_proc () #4 0x0000003720e07893 in start_thread () from /lib64/libpthread.so.0 #5 0x00000037206f4bfd in clone () from /lib64/libc.so.6 Thread 1 (Thread 0x7f60a610c700 (LWP 87746)): #0 0x0000003720e080e5 in pthread_join () from /lib64/libpthread.so.0 #1 0x0000000000400806 in main () .... $ pstack 87746 Thread 3 (Thread 0x7f60a610a700 (LWP 87747)): #0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 #1 0x0000003720e093ca in _L_lock_829 () from /lib64/libpthread.so.0 #2 0x0000003720e09298 in pthread_mutex_lock () from /lib64/libpthread.so.0 #3 0x0000000000400725 in threadA_proc () #4 0x0000003720e07893 in start_thread () from /lib64/libpthread.so.0 #5 0x00000037206f4bfd in clone () from /lib64/libc.so.6 Thread 2 (Thread 0x7f60a5709700 (LWP 87748)): #0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 #1 0x0000003720e093ca in _L_lock_829 () from /lib64/libpthread.so.0 #2 0x0000003720e09298 in pthread_mutex_lock () from /lib64/libpthread.so.0 #3 0x0000000000400792 in threadB_proc () #4 0x0000003720e07893 in start_thread () from /lib64/libpthread.so.0 #5 0x00000037206f4bfd in clone () from /lib64/libc.so.6 Thread 1 (Thread 0x7f60a610c700 (LWP 87746)): #0 0x0000003720e080e5 in pthread_join () from /lib64/libpthread.so.0 #1 0x0000000000400806 in main ()可以看到，Thread 2 和 Thread 3 一直阻塞获取锁（pthread_mutex_lock）的过程，而且 pstack 多次输出信息都没有变化，那么可能大概率发生了死锁。但是，还不能够确认这两个线程是在互相等待对方的锁的释放，因为我们看不到它们是等在哪个锁对象，于是我们可以使用 gdb 工具进一步确认。整个 gdb 调试过程，如下：// gdb 命令 $ gdb -p 87746 // 打印所有的线程信息 (gdb) info thread 3 Thread 0x7f60a610a700 (LWP 87747) 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 2 Thread 0x7f60a5709700 (LWP 87748) 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 * 1 Thread 0x7f60a610c700 (LWP 87746) 0x0000003720e080e5 in pthread_join () from /lib64/libpthread.so.0 //最左边的 * 表示 gdb 锁定的线程，切换到第二个线程去查看 // 切换到第2个线程 (gdb) thread 2 [Switching to thread 2 (Thread 0x7f60a5709700 (LWP 87748))]#0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 // bt 可以打印函数堆栈，却无法看到函数参数，跟 pstack 命令一样 (gdb) bt #0 0x0000003720e0da1d in __lll_lock_wait () from /lib64/libpthread.so.0 #1 0x0000003720e093ca in _L_lock_829 () from /lib64/libpthread.so.0 #2 0x0000003720e09298 in pthread_mutex_lock () from /lib64/libpthread.so.0 #3 0x0000000000400792 in threadB_proc (data=0x0) at dead_lock.c:25 #4 0x0000003720e07893 in start_thread () from /lib64/libpthread.so.0 #5 0x00000037206f4bfd in clone () from /lib64/libc.so.6 // 打印第三帧信息，每次函数调用都会有压栈的过程，而 frame 则记录栈中的帧信息 (gdb) frame 3 #3 0x0000000000400792 in threadB_proc (data=0x0) at dead_lock.c:25 27 printf("thread B waiting get ResourceA \n"); 28 pthread_mutex_lock(&mutex_A); // 打印mutex_A的值 , __owner表示gdb中标示线程的值，即LWP (gdb) p mutex_A $1 = {__data = {__lock = 2, __count = 0, __owner = 87747, __nusers = 1, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = "\002\000\000\000\000\000\000\000\303V\001\000\001", '\000' <repeats 26 times>, __align = 2} // 打印mutex_B的值 , __owner表示gdb中标示线程的值，即LWP (gdb) p mutex_B $2 = {__data = {__lock = 2, __count = 0, __owner = 87748, __nusers = 1, __kind = 0, __spins = 0, __list = {__prev = 0x0, __next = 0x0}}, __size = "\002\000\000\000\000\000\000\000\304V\001\000\001", '\000' <repeats 26 times>, __align = 2} 我来解释下，上面的调试过程：通过 info thread 打印了所有的线程信息，可以看到有 3 个线程，一个是主线程（LWP 87746），另外两个都是我们自己创建的线程（LWP 87747 和 87748）；通过 thread 2，将切换到第 2 个线程（LWP 87748）；通过 bt，打印线程的调用栈信息，可以看到有 threadB_proc 函数，说明这个是线程 B 函数，也就说 LWP 87748 是线程 B;通过 frame 3，打印调用栈中的第三个帧的信息，可以看到线程 B 函数，在获取互斥锁 A 的时候阻塞了；通过 p mutex_A，打印互斥锁 A 对象信息，可以看到它被 LWP 为 87747（线程 A）的线程持有着；通过 p mutex_B，打印互斥锁 B 对象信息，可以看到他被 LWP 为 87748（线程 B）的线程持有着；因为线程 B 在等待线程 A 所持有的 mutex_A, 而同时线程 A 又在等待线程 B 所拥有的 mutex_B, 所以可以断定该程序发生了死锁。避免死锁问题的发生前面我们提到，产生死锁的四个必要条件是：互斥条件、持有并等待条件、不可剥夺条件、环路等待条件。那么避免死锁问题就只需要破环其中一个条件就可以，最常见的并且可行的就是使用资源有序分配法，来破环环路等待条件。那什么是资源有序分配法呢？线程 A 和 线程 B 获取资源的顺序要一样，当线程 A 是先尝试获取资源 A，然后尝试获取资源 B 的时候，线程 B 同样也是先尝试获取资源 A，然后尝试获取资源 B。也就是说，线程 A 和 线程 B 总是以相同的顺序申请自己想要的资源。我们使用资源有序分配法的方式来修改前面发生死锁的代码，我们可以不改动线程 A 的代码。我们先要清楚线程 A 获取资源的顺序，它是先获取互斥锁 A，然后获取互斥锁 B。所以我们只需将线程 B 改成以相同顺序的获取资源，就可以打破死锁了。线程 B 函数改进后的代码如下：//线程 B 函数，同线程 A 一样，先获取互斥锁 A，然后获取互斥锁 B void *threadB_proc(void *data) { printf("thread B waiting get ResourceA \n"); pthread_mutex_lock(&mutex_A); printf("thread B got ResourceA \n"); sleep(1); printf("thread B waiting get ResourceB \n"); pthread_mutex_lock(&mutex_B); printf("thread B got ResourceB \n"); pthread_mutex_unlock(&mutex_B); pthread_mutex_unlock(&mutex_A); return (void *)0; }执行结果如下，可以看，没有发生死锁。thread B waiting get ResourceA thread B got ResourceA thread A waiting get ResourceA thread B waiting get ResourceB thread B got ResourceB thread A got ResourceA thread A waiting get ResourceB thread A got ResourceB exit总结简单来说，死锁问题的产生是由两个或者以上线程并行执行的时候，争夺资源而互相等待造成的。死锁只有同时满足互斥、持有并等待、不可剥夺、环路等待这四个条件的时候才会发生。所以要避免死锁问题，就是要破坏其中一个条件即可，最常用的方法就是使用资源有序分配法来破坏环路等待条件。
AI 摘要 文章通过实战演示 Java Socket 通信：服务端用 ServerSocket 监听，客户端用 Socket 连接，双方以对象流收发 Message。指出单线程只能处理一个连接的缺陷，提出线程池改进，并说明 BIO 模型局限，引出后续用 Netty 实现 NIO 的方案。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结05 Socket 网络通信实战05 Socket 网络通信实战前言guide-rpc-framework 的第一版使用的是 JDK 提供了 socket 进行网络编程。为了搞懂具体原理，我们首先要学会使用 Socket 进行网络通信。什么是 Socket(套接字)Socket 是一个抽象概念，应用程序可以通过它发送或接收数据。在使用 Socket 进行网络通信的时候，通过 Socket 就可以让我们的数据在网络中传输。操作套接字的时候，和我们读写文件很像。套接字是 IP 地址与端口的组合，套接字 Socket=（IP 地址：端口号）。要通过互联网进行通信，至少需要一对套接字：运行于服务器端的 Server Socket。运行于客户机端的 Client Socket在 Java 开发中使用 Socket 时会常用到两个类，都在 java.net 包中：Socket: 一般用于客户端ServerSocket :用于服务端Socket 网络通信过程Socket 网络通信过程如下图所示：https://www.javatpoint.com/socket-programmingSocket 网络通信过程简单来说分为下面 4 步：建立服务端并且监听客户端请求客户端请求，服务端和客户端建立连接两端之间可以传递数据关闭资源对应到服务端和客户端的话，是下面这样的。服务器端：创建 ServerSocket 对象并且绑定地址（ip）和端口号(port)：server.bind(new InetSocketAddress(host, port))通过 accept()方法监听客户端请求连接建立后，通过输入流读取客户端发送的请求信息通过输出流向客户端发送响应信息关闭相关资源客户端：创建Socket 对象并且连接指定的服务器的地址（ip）和端口号(port)：socket.connect(inetSocketAddress)连接建立后，通过输出流向服务器端发送请求信息通过输入流获取服务器响应的信息关闭相关资源Socket 网络通信实战服务端public class HelloServer { private static final Logger logger = LoggerFactory.getLogger(HelloServer.class); public void start(int port) { //1.创建 ServerSocket 对象并且绑定一个端口 try (ServerSocket server = new ServerSocket(port);) { Socket socket; //2.通过 accept()方法监听客户端请求 while ((socket = server.accept()) != null) { logger.info("client connected"); try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) { //3.通过输入流读取客户端发送的请求信息 Message message = (Message) objectInputStream.readObject(); logger.info("server receive message:" + message.getContent()); message.setContent("new content"); //4.通过输出流向客户端发送响应信息 objectOutputStream.writeObject(message); objectOutputStream.flush(); } catch (IOException | ClassNotFoundException e) { logger.error("occur exception:", e); } } } catch (IOException e) { logger.error("occur IOException:", e); } } public static void main(String[] args) { HelloServer helloServer = new HelloServer(); helloServer.start(6666); } }ServerSocket 的 accept（） 方法是阻塞方法，也就是说 ServerSocket 在调用 accept（)等待客户端的连接请求时会阻塞，直到收到客户端发送的连接请求才会继续往下执行代码。很明显，我上面演示的代码片段有一个很严重的问题：只能同时处理一个客户端的连接，如果需要管理多个客户端的话，就需要为我们请求的客户端单独创建一个线程。 如下图所示：对应的 Java 代码可能是下面这样的：new Thread(() -> { // 创建 socket 连接 }).start();但是，这样会导致一个很严重的问题：资源浪费。我们知道线程是很宝贵的资源，如果我们为每一次连接都用一个线程处理的话，就会导致线程越来越多，最后达到了极限之后，就无法再创建线程处理请求了。处理的不好的话，甚至可能直接就宕机掉了。很多人就会问了：那有没有改进的方法呢？当然有！ 比较简单并且实际的改进方法就是使用线程池。线程池还可以让线程的创建和回收成本相对较低，并且我们可以指定线程池的可创建线程的最大数量，这样就不会导致线程创建过多，机器资源被不合理消耗。ThreadFactory threadFactory = Executors.defaultThreadFactory(); ExecutorService threadPool = new ThreadPoolExecutor(10, 100, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100), threadFactory); threadPool.execute(() -> { // 创建 socket 连接 });但是，即使你再怎么优化和改变。也改变不了它的底层仍然是同步阻塞的 BIO 模型的事实，因此无法从根本上解决问题。为了解决上述的问题，Java 1.4 中引入了 NIO ，一种同步非阻塞的 I/O 模型。 由于使用同步非阻塞的 I/O 模型 NIO 来进行网络编程真的太麻烦了。你可以使用基于 NIO 的网络编程框架 Netty ，它将是你最好的选择（前面的章节提到过，后面的章节会详细讲解如何使用 Netty 进行网络编程）！客户端public class HelloClient { private static final Logger logger = LoggerFactory.getLogger(HelloClient.class); public Object send(Message message, String host, int port) { //1. 创建Socket对象并且指定服务器的地址和端口号 try (Socket socket = new Socket(host, port)) { ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); //2.通过输出流向服务器端发送请求信息 objectOutputStream.writeObject(message); //3.通过输入流获取服务器响应的信息 ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); return objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { logger.error("occur exception:", e); } return null; } public static void main(String[] args) { HelloClient helloClient = new HelloClient(); Message message = (Message) helloClient.send(new Message("content from client"), "127.0.0.1", 6666); System.out.println("client receive message:" + message.getContent()); } }发送的消息实体类：@Data @AllArgsConstructor public class Message implements Serializable { private String content; }首先运行服务端，然后再运行客户端，控制台输出如下：服务端:[main] INFO github.javaguide.socket.HelloServer - client connected [main] INFO github.javaguide.socket.HelloServer - server receive message:content from client客户端：client receive message:new content好的！我们的第一个使用 Socket 进行网络编程的案例已经完成了。下一篇我们来看看如何使用 Netty 进行网络编程。
AI 摘要 Redis主从复制通过全量复制、长连接命令传播和增量复制三阶段实现：首次同步用RDB+缓冲写命令，后续维持TCP长连接实时传播写命令；断网重连后按repl_backlog_buffer环形缓存做增量同步，若缓存不足则退为全量同步。可设“经理”从节点分摊RDB压力，并通过哨兵完成故障自动切换。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结主从复制是怎么实现的？大家好，我是樱舞。我在前两篇已经给大家图解了 AOF 和 RDB，这两个持久化技术保证了即使在服务器重启的情况下也不会丢失数据（或少量损失）。不过，由于数据都是存储在一台服务器上，如果出事就完犊子了，比如：如果服务器发生了宕机，由于数据恢复是需要点时间，那么这个期间是无法服务新的请求的；如果这台服务器的硬盘出现了故障，可能数据就都丢失了。要避免这种单点故障，最好的办法是将数据备份到其他服务器上，让这些服务器也可以对外提供服务，这样即使有一台服务器出现了故障，其他服务器依然可以继续提供服务。多台服务器要保存同一份数据，这里问题就来了。这些服务器之间的数据如何保持一致性呢？数据的读写操作是否每台服务器都可以处理？Redis 提供了主从复制模式，来避免上述的问题。这个模式可以保证多台服务器的数据一致性，且主从服务器之间采用的是「读写分离」的方式。主服务器可以进行读写操作，当发生写操作时自动将写操作同步给从服务器，而从服务器一般是只读，并接受主服务器同步过来写操作命令，然后执行这条命令。也就是说，所有的数据修改只在主服务器上进行，然后将最新的数据同步给从服务器，这样就使得主从服务器的数据是一致的。同步这两个字说的简单，但是这个同步过程并没有想象中那么简单，要考虑的事情不是一两个。我们先来看看，主从服务器间的第一次同步是如何工作的？第一次同步多台服务器之间要通过什么方式来确定谁是主服务器，或者谁是从服务器呢？我们可以使用 replicaof（Redis 5.0 之前使用 slaveof）命令形成主服务器和从服务器的关系。比如，现在有服务器 A 和 服务器 B，我们在服务器 B 上执行下面这条命令：# 服务器 B 执行这条命令 replicaof <服务器 A 的 IP 地址> <服务器 A 的 Redis 端口号>接着，服务器 B 就会变成服务器 A 的「从服务器」，然后与主服务器进行第一次同步。主从服务器间的第一次同步的过程可分为三个阶段：第一阶段是建立链接、协商同步；第二阶段是主服务器同步数据给从服务器；第三阶段是主服务器发送新写操作命令给从服务器。为了让你更清楚了解这三个阶段，我画了一张图。接下来，我在具体介绍每一个阶段都做了什么。第一阶段：建立链接、协商同步执行了 replicaof 命令后，从服务器就会给主服务器发送 psync 命令，表示要进行数据同步。psync 命令包含两个参数，分别是主服务器的 runID 和复制进度 offset。runID，每个 Redis 服务器在启动时都会自动生产一个随机的 ID 来唯一标识自己。当从服务器和主服务器第一次同步时，因为不知道主服务器的 run ID，所以将其设置为 "?"。offset，表示复制的进度，第一次同步时，其值为 -1。主服务器收到 psync 命令后，会用 FULLRESYNC 作为响应命令返回给对方。并且这个响应命令会带上两个参数：主服务器的 runID 和主服务器目前的复制进度 offset。从服务器收到响应后，会记录这两个值。FULLRESYNC 响应命令的意图是采用全量复制的方式，也就是主服务器会把所有的数据都同步给从服务器。所以，第一阶段的工作时为了全量复制做准备。那具体怎么全量同步呀呢？我们可以往下看第二阶段。第二阶段：主服务器同步数据给从服务器接着，主服务器会执行 bgsave 命令来生成 RDB 文件，然后把文件发送给从服务器。从服务器收到 RDB 文件后，会先清空当前的数据，然后载入 RDB 文件。这里有一点要注意，主服务器生成 RDB 这个过程是不会阻塞主线程的，因为 bgsave 命令是产生了一个子进程来做生成 RDB 文件的工作，是异步工作的，这样 Redis 依然可以正常处理命令。但是，这期间的写操作命令并没有记录到刚刚生成的 RDB 文件中，这时主从服务器间的数据就不一致了。那么为了保证主从服务器的数据一致性，主服务器在下面这三个时间间隙中将收到的写操作命令，写入到 replication buffer 缓冲区里：主服务器生成 RDB 文件期间；主服务器发送 RDB 文件给从服务器期间；「从服务器」加载 RDB 文件期间；第三阶段：主服务器发送新写操作命令给从服务器在主服务器生成的 RDB 文件发送完，从服务器收到 RDB 文件后，丢弃所有旧数据，将 RDB 数据载入到内存。完成 RDB 的载入后，会回复一个确认消息给主服务器。接着，主服务器将 replication buffer 缓冲区里所记录的写操作命令发送给从服务器，从服务器执行来自主服务器 replication buffer 缓冲区里发来的命令，这时主从服务器的数据就一致了。至此，主从服务器的第一次同步的工作就完成了。命令传播主从服务器在完成第一次同步后，双方之间就会维护一个 TCP 连接。后续主服务器可以通过这个连接继续将写操作命令传播给从服务器，然后从服务器执行该命令，使得与主服务器的数据库状态相同。而且这个连接是长连接的，目的是避免频繁的 TCP 连接和断开带来的性能开销。上面的这个过程被称为基于长连接的命令传播，通过这种方式来保证第一次同步后的主从服务器的数据一致性。分摊主服务器的压力在前面的分析中，我们可以知道主从服务器在第一次数据同步的过程中，主服务器会做两件耗时的操作：生成 RDB 文件和传输 RDB 文件。主服务器是可以有多个从服务器的，如果从服务器数量非常多，而且都与主服务器进行全量同步的话，就会带来两个问题：由于是通过 bgsave 命令来生成 RDB 文件的，那么主服务器就会忙于使用 fork() 创建子进程，如果主服务器的内存数据非常大，在执行 fork() 函数时是会阻塞主线程的，从而使得 Redis 无法正常处理请求；传输 RDB 文件会占用主服务器的网络带宽，会对主服务器响应命令请求产生影响。这种情况就好像，刚创业的公司，由于人不多，所以员工都归老板一个人管，但是随着公司的发展，人员的扩充，老板慢慢就无法承担全部员工的管理工作了。要解决这个问题，老板就需要设立经理职位，由经理管理多名普通员工，然后老板只需要管理经理就好。Redis 也是一样的，从服务器可以有自己的从服务器，我们可以把拥有从服务器的从服务器当作经理角色，它不仅可以接收主服务器的同步数据，自己也可以同时作为主服务器的形式将数据同步给从服务器，组织形式如下图：通过这种方式，主服务器生成 RDB 和传输 RDB 的压力可以分摊到充当经理角色的从服务器。那具体怎么做到的呢？其实很简单，我们在「从服务器」上执行下面这条命令，使其作为目标服务器的从服务器：replicaof <目标服务器的 IP> 6379此时如果目标服务器本身也是「从服务器」，那么该目标服务器就会成为「经理」的角色，不仅可以接受主服务器同步的数据，也会把数据同步给自己旗下的从服务器，从而减轻主服务器的负担。增量复制主从服务器在完成第一次同步后，就会基于长连接进行命令传播。可是，网络总是不按套路出牌的嘛，说延迟就延迟，说断开就断开。如果主从服务器间的网络连接断开了，那么就无法进行命令传播了，这时从服务器的数据就没办法和主服务器保持一致了，客户端就可能从「从服务器」读到旧的数据。那么问题来了，如果此时断开的网络，又恢复正常了，要怎么继续保证主从服务器的数据一致性呢？在 Redis 2.8 之前，如果主从服务器在命令同步时出现了网络断开又恢复的情况，从服务器就会和主服务器重新进行一次全量复制，很明显这样的开销太大了，必须要改进一波。所以，从 Redis 2.8 开始，网络断开又恢复后，从主从服务器会采用增量复制的方式继续同步，也就是只会把网络断开期间主服务器接收到的写操作命令，同步给从服务器。网络恢复后的增量复制过程如下图：主要有三个步骤：从服务器在恢复网络后，会发送 psync 命令给主服务器，此时的 psync 命令里的 offset 参数不是 -1；主服务器收到该命令后，然后用 CONTINUE 响应命令告诉从服务器接下来采用增量复制的方式同步数据；然后主服务将主从服务器断线期间，所执行的写命令发送给从服务器，然后从服务器执行这些命令。那么关键的问题来了，主服务器怎么知道要将哪些增量数据发送给从服务器呢？答案藏在这两个东西里：repl_backlog_buffer，是一个「环形」缓冲区，用于主从服务器断连后，从中找到差异的数据；replication offset，标记上面那个缓冲区的同步进度，主从服务器都有各自的偏移量，主服务器使用 master_repl_offset 来记录自己「写」到的位置，从服务器使用 slave_repl_offset 来记录自己「读」到的位置。那 repl_backlog_buffer 缓冲区是什么时候写入的呢？在主服务器进行命令传播时，不仅会将写命令发送给从服务器，还会将写命令写入到 repl_backlog_buffer 缓冲区里，因此 这个缓冲区里会保存着最近传播的写命令。网络断开后，当从服务器重新连上主服务器时，从服务器会通过 psync 命令将自己的复制偏移量 slave_repl_offset 发送给主服务器，主服务器根据自己的 master_repl_offset 和 slave_repl_offset 之间的差距，然后来决定对从服务器执行哪种同步操作：如果判断出从服务器要读取的数据还在 repl_backlog_buffer 缓冲区里，那么主服务器将采用增量同步的方式；相反，如果判断出从服务器要读取的数据已经不存在 repl_backlog_buffer 缓冲区里，那么主服务器将采用全量同步的方式。当主服务器在 repl_backlog_buffer 中找到主从服务器差异（增量）的数据后，就会将增量的数据写入到 replication buffer 缓冲区，这个缓冲区我们前面也提到过，它是缓存将要传播给从服务器的命令。repl_backlog_buffer 缓行缓冲区的默认大小是 1M，并且由于它是一个环形缓冲区，所以当缓冲区写满后，主服务器继续写入的话，就会覆盖之前的数据。因此，当主服务器的写入速度远超于从服务器的读取速度，缓冲区的数据一下就会被覆盖。那么在网络恢复时，如果从服务器想读的数据已经被覆盖了，主服务器就会采用全量同步，这个方式比增量同步的性能损耗要大很多。因此，为了避免在网络恢复时，主服务器频繁地使用全量同步的方式，我们应该调整下 repl_backlog_buffer 缓冲区大小，尽可能的大一些，减少出现从服务器要读取的数据被覆盖的概率，从而使得主服务器采用增量同步的方式。那 repl_backlog_buffer 缓冲区具体要调整到多大呢？repl_backlog_buffer 最小的大小可以根据这面这个公式估算。我来解释下这个公式的意思：second 为从服务器断线后重新连接上主服务器所需的平均 时间 (以秒计算)。write_size_per_second 则是主服务器平均每秒产生的写命令数据量大小。举个例子，如果主服务器平均每秒产生 1 MB 的写命令，而从服务器断线之后平均要 5 秒才能重新连接主服务器。那么 repl_backlog_buffer 大小就不能低于 5 MB，否则新写地命令就会覆盖旧数据了。当然，为了应对一些突发的情况，可以将 repl_backlog_buffer 的大小设置为此基础上的 2 倍，也就是 10 MB。关于 repl_backlog_buffer 大小修改的方法，只需要修改配置文件里下面这个参数项的值就可以。repl-backlog-size 1mb总结主从复制共有三种模式：全量复制、基于长连接的命令传播、增量复制。主从服务器第一次同步的时候，就是采用全量复制，此时主服务器会有两个耗时的地方，分别是生成 RDB 文件和传输 RDB 文件。为了避免过多的从服务器和主服务器进行全量复制，可以把一部分从服务器升级为「经理角色」，让它也有自己的从服务器，通过这样可以分摊主服务器的压力。第一次同步完成后，主从服务器都会维护着一个长连接，主服务器在接收到写操作命令后，就会通过这个连接将写命令传播给从服务器，来保证主从服务器的数据一致性。如果遇到网络断开，增量复制就可以上场了，不过这个还跟 repl_backlog_size 这个大小有关系。如果它配置的过小，主从服务器网络恢复时，可能发生「从服务器」想读的数据已经被覆盖了，那么这时就会导致主服务器采用全量复制的方式。所以为了避免这种情况的频繁发生，要调大这个参数的值，以降低主从服务器断开后全量同步的概率。面试题Redis 主从节点是长连接还是短链接？长连接怎么判断 Redis 某个节点是否正常工作？Redis 判断节点是否正常工作，基本都是通过互相的 ping-pong 心跳检测机制，如果有一半以上的节点去 ping 一个节点的时候没有 pong 回应，集群就会认为这个节点挂掉了，会断开与这个节点的连接。Redis 主从节点发送的心跳间隔是不一样的，而且作用也有一点区别：Redis 主节点默认每隔 10 秒对从节点发送 ping 命令，判断从节点的存活性和连接状态，可通过参数 repl-ping-slave-period 控制发送频率。Redis 从节点每隔 1 秒发送 replconf ack{offset} 命令，给主节点上报自身当前的复制偏移量，目的是为了：实时监测主从节点网络状态；上报自身复制偏移量，检查复制数据是否丢失，如果从节点数据丢失，再从主节点的复制缓冲区中拉取丢失数据。主从复制架构中，过期 key 如何处理？主节点处理了一个 key 或者通过淘汰算法淘汰了一个 key，这个时间主节点模拟一条 del 命令发送给从节点，从节点收到该命令后，就进行删除 key 的操作。Redis 是同步复制还是异步复制？Redis 主节点每次收到写命令之后，先写到内部的缓冲区，然后异步发送给从节点。主从复制中两个 Buffer(replication buffer、repl backlog buffer) 有什么区别？replication buffer、repl backlog buffer 区别如下：出现的阶段不一样：repl backlog buffer 是在增量复制阶段出现，一个主节点只分配一个 repl backlog buffer；replication buffer 是在全量复制阶段和增量复制阶段都会出现，主节点会给每个新连接的从节点，分配一个 replication buffer；这两个 Buffer 都有大小限制的，当缓冲区满了之后，发生的事情不一样：当 repl backlog buffer 满了，因为是环形结构，会直接覆盖起始位置数据;当 replication buffer 满了，会导致连接断开，删除缓存，从节点重新连接，重新开始全量复制。如何应对主从数据不一致？为什么会出现主从数据不一致？主从数据不一致，就是指客户端从从节点中读取到的值和主节点中的最新值并不一致。之所以会出现主从数据不一致的现象，是因为主从节点间的命令复制是异步进行的，所以无法实现强一致性保证（主从数据时时刻刻保持一致）。具体来说，在主从节点命令传播阶段，主节点收到新的写命令后，会发送给从节点。但是，主节点并不会等到从节点实际执行完命令后，再把结果返回给客户端，而是主节点自己在本地执行完命令后，就会向客户端返回结果了。如果从节点还没有执行主节点同步过来的命令，主从节点间的数据就不一致了。如何如何应对主从数据不一致？第一种方法，尽量保证主从节点间的网络连接状况良好，避免主从节点在不同的机房。第二种方法，可以开发一个外部程序来监控主从节点间的复制进度。具体做法：Redis 的 INFO replication 命令可以查看主节点接收写命令的进度信息（master_repl_offset）和从节点复制写命令的进度信息（slave_repl_offset），所以，我们就可以开发一个监控程序，先用 INFO replication 命令查到主、从节点的进度，然后，我们用 master_repl_offset 减去 slave_repl_offset，这样就能得到从节点和主节点间的复制进度差值了。如果某个从节点的进度差值大于我们预设的阈值，我们可以让客户端不再和这个从节点连接进行数据读取，这样就可以减少读到不一致数据的情况。不过，为了避免出现客户端和所有从节点都不能连接的情况，我们需要把复制进度差值的阈值设置得大一些。主从切换如何减少数据丢失？主从切换过程中，产生数据丢失的情况有两种：异步复制同步丢失集群产生脑裂数据丢失我们不可能保证数据完全不丢失，只能做到使得尽量少的数据丢失。异步复制同步丢失对于 Redis 主节点与从节点之间的数据复制，是异步复制的，当客户端发送写请求给主节点的时候，客户端会返回 ok，接着主节点将写请求异步同步给各个从节点，但是如果此时主节点还没来得及同步给从节点时发生了断电，那么主节点内存中的数据会丢失。减少异步复制的数据丢失的方案Redis 配置里有一个参数 min-slaves-max-lag，表示一旦所有的从节点数据复制和同步的延迟都超过了 min-slaves-max-lag 定义的值，那么主节点就会拒绝接收任何请求。假设将 min-slaves-max-lag 配置为 10s 后，根据目前 master->slave 的复制速度，如果数据同步完成所需要时间超过 10s，就会认为 master 未来宕机后损失的数据会很多，master 就拒绝写入新请求。这样就能将 master 和 slave 数据差控制在 10s 内，即使 master 宕机也只是这未复制的 10s 数据。那么对于客户端，当客户端发现 master 不可写后，我们可以采取降级措施，将数据暂时写入本地缓存和磁盘中，在一段时间（等 master 恢复正常）后重新写入 master 来保证数据不丢失，也可以将数据写入 kafka 消息队列，等 master 恢复正常，再隔一段时间去消费 kafka 中的数据，让将数据重新写入 master。集群产生脑裂数据丢失先来理解集群的脑裂现象，这就好比一个人有两个大脑，那么到底受谁控制呢？那么在 Redis 中，集群脑裂产生数据丢失的现象是怎样的呢？在 Redis 主从架构中，部署方式一般是「一主多从」，主节点提供写操作，从节点提供读操作。如果主节点的网络突然发生了问题，它与所有的从节点都失联了，但是此时的主节点和客户端的网络是正常的，这个客户端并不知道 Redis 内部已经出现了问题，还在照样的向这个失联的主节点写数据（过程 A），此时这些数据被主节点缓存到了缓冲区里，因为主从节点之间的网络问题，这些数据都是无法同步给从节点的。这时，哨兵也发现主节点失联了，它就认为主节点挂了（但实际上主节点正常运行，只是网络出问题了），于是哨兵就会在从节点中选举出一个 leader 作为主节点，这时集群就有两个主节点了 —— 脑裂出现了。这时候网络突然好了，哨兵因为之前已经选举出一个新主节点了，它就会把旧主节点降级为从节点（A），然后从节点（A）会向新主节点请求数据同步，因为第一次同步是全量同步的方式，此时的从节点（A）会清空掉自己本地的数据，然后再做全量同步。所以，之前客户端在过程 A 写入的数据就会丢失了，也就是集群产生脑裂数据丢失的问题。总结一句话就是：由于网络问题，集群节点之间失去联系。主从数据不同步；重新平衡选举，产生两个主服务。等网络恢复，旧主节点会降级为从节点，再与新主节点进行同步复制的时候，由于会从节点会清空自己的缓冲区，所以导致之前客户端写入的数据丢失了。减少脑裂的数据丢的方案当主节点发现「从节点下线的数量太多」，或者「网络延迟太大」的时候，那么主节点会禁止写操作，直接把错误返回给客户端。在 Redis 的配置文件中有两个参数我们可以设置：min-slaves-to-write x，主节点必须要有至少 x 个从节点连接，如果小于这个数，主节点会禁止写数据。min-slaves-max-lag x，主从数据复制和同步的延迟不能超过 x 秒，如果超过，主节点会禁止写数据。我们可以把 min-slaves-to-write 和 min-slaves-max-lag 这两个配置项搭配起来使用，分别给它们设置一定的阈值，假设为 N 和 T。这两个配置项组合后的要求是，主节点连接的从节点中至少有 N 个从节点，「并且」主节点进行数据复制时的 ACK 消息延迟不能超过 T 秒，否则，主节点就不会再接收客户端的写请求了。即使原主节点是假故障，它在假故障期间也无法响应哨兵心跳，也不能和从节点进行同步，自然也就无法和从节点进行 ACK 确认了。这样一来，min-slaves-to-write 和 min-slaves-max-lag 的组合要求就无法得到满足，原主节点就会被限制接收客户端写请求，客户端也就不能在原主节点中写入新数据了。等到新主节点上线时，就只有新主节点能接收和处理客户端请求，此时，新写的数据会被直接写到新主节点中。而原主节点会被哨兵降为从节点，即使它的数据被清空了，也不会有新数据丢失。我再来给你举个例子。假设我们将 min-slaves-to-write 设置为 1，把 min-slaves-max-lag 设置为 12s，把哨兵的 down-after-milliseconds 设置为 10s，主节点因为某些原因卡住了 15s，导致哨兵判断主节点客观下线，开始进行主从切换。同时，因为原主节点卡住了 15s，没有一个从节点能和原主节点在 12s 内进行数据复制，原主节点也无法接收客户端请求了。这样一来，主从切换完成后，也只有新主节点能接收请求，不会发生脑裂，也就不会发生数据丢失的问题了。主从如何做到故障自动切换？主节点挂了，从节点是无法自动升级为主节点的，这个过程需要人工处理，在此期间 Redis 无法对外提供写操作。此时，Redis 哨兵机制就登场了，哨兵在发现主节点出现故障时，由哨兵自动完成故障发现和故障转移，并通知给应用方，从而实现高可用性。
什么是高可用？可用性的判断标准是啥？
AI 摘要 高可用指系统长期持续可服务，用“几个9”衡量，如99.9999%。提升方法：严把代码质量、集群化、限流、超时重试、熔断、异步、缓存、灰度发布、监控报警、定期换硬件等。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: 高可用系统设计指南category: 高可用icon: design什么是高可用？可用性的判断标准是啥？高可用描述的是一个系统在大部分时间都是可用的，可以为我们提供服务的。高可用代表系统即使在发生硬件故障或者系统升级的时候，服务仍然是可用的。一般情况下，我们使用多少个 9 来评判一个系统的可用性，比如 99.9999% 就是代表该系统在所有的运行时间中只有 0.0001% 的时间是不可用的，这样的系统就是非常非常高可用的了！当然，也会有系统如果可用性不太好的话，可能连 9 都上不了。除此之外，系统的可用性还可以用某功能的失败次数与总的请求次数之比来衡量，比如对网站请求 1000 次，其中有 10 次请求失败，那么可用性就是 99%。哪些情况会导致系统不可用？黑客攻击；硬件故障，比如服务器坏掉。并发量/用户请求量激增导致整个服务宕掉或者部分服务不可用。代码中的坏味道导致内存泄漏或者其他问题导致程序挂掉。网站架构某个重要的角色比如 Nginx 或者数据库突然不可用。自然灾害或者人为破坏。……有哪些提高系统可用性的方法？注重代码质量，测试严格把关我觉得这个是最最最重要的，代码质量有问题比如比较常见的内存泄漏、循环依赖都是对系统可用性极大的损害。大家都喜欢谈限流、降级、熔断，但是我觉得从代码质量这个源头把关是首先要做好的一件很重要的事情。如何提高代码质量？比较实际可用的就是 CodeReview，不要在乎每天多花的那 1 个小时左右的时间，作用可大着呢！另外，安利几个对提高代码质量有实际效果的神器：Sonarqube；Alibaba 开源的 Java 诊断工具 Arthas；阿里巴巴 Java 代码规范（Alibaba Java Code Guidelines）；IDEA 自带的代码分析等工具。使用集群，减少单点故障先拿常用的 Redis 举个例子！我们如何保证我们的 Redis 缓存高可用呢？答案就是使用集群，避免单点故障。当我们使用一个 Redis 实例作为缓存的时候，这个 Redis 实例挂了之后，整个缓存服务可能就挂了。使用了集群之后，即使一台 Redis 实例挂了，不到一秒就会有另外一台 Redis 实例顶上。限流流量控制（flow control），其原理是监控应用流量的 QPS 或并发线程数等指标，当达到指定的阈值时对流量进行控制，以避免被瞬时的流量高峰冲垮，从而保障应用的高可用性。——来自 alibaba-Sentinel 的 wiki。超时和重试机制设置一旦用户请求超过某个时间的得不到响应，就抛出异常。这个是非常重要的，很多线上系统故障都是因为没有进行超时设置或者超时设置的方式不对导致的。我们在读取第三方服务的时候，尤其适合设置超时和重试机制。一般我们使用一些 RPC 框架的时候，这些框架都自带的超时重试的配置。如果不进行超时设置可能会导致请求响应速度慢，甚至导致请求堆积进而让系统无法再处理请求。重试的次数一般设为 3 次，再多次的重试没有好处，反而会加重服务器压力（部分场景使用失败重试机制会不太适合）。熔断机制超时和重试机制设置之外，熔断机制也是很重要的。 熔断机制说的是系统自动收集所依赖服务的资源使用情况和性能指标，当所依赖的服务恶化或者调用失败次数达到某个阈值的时候就迅速失败，让当前系统立即切换依赖其他备用服务。 比较常用的流量控制和熔断降级框架是 Netflix 的 Hystrix 和 alibaba 的 Sentinel。异步调用异步调用的话我们不需要关心最后的结果，这样我们就可以用户请求完成之后就立即返回结果，具体处理我们可以后续再做，秒杀场景用这个还是蛮多的。但是，使用异步之后我们可能需要 适当修改业务流程进行配合，比如用户在提交订单之后，不能立即返回用户订单提交成功，需要在消息队列的订单消费者进程真正处理完该订单之后，甚至出库后，再通过电子邮件或短信通知用户订单成功。除了可以在程序中实现异步之外，我们常常还使用消息队列，消息队列可以通过异步处理提高系统性能（削峰、减少响应所需时间）并且可以降低系统耦合性。使用缓存如果我们的系统属于并发量比较高的话，如果我们单纯使用数据库的话，当大量请求直接落到数据库可能数据库就会直接挂掉。使用缓存缓存热点数据，因为缓存存储在内存中，所以速度相当地快！其他核心应用和服务优先使用更好的硬件监控系统资源使用情况增加报警设置。注意备份，必要时候回滚。灰度发布： 将服务器集群分成若干部分，每天只发布一部分机器，观察运行稳定没有故障，第二天继续发布一部分机器，持续几天才把整个集群全部发布完毕，期间如果发现问题，只需要回滚已发布的一部分服务器即可定期检查/更换硬件： 如果不是购买的云服务的话，定期还是需要对硬件进行一波检查的，对于一些需要更换或者升级的硬件，要及时更换或者升级。……
AI 摘要 文章系统梳理Java线程池：从池化价值、Executor框架、ThreadPoolExecutor七大参数与饱和策略，到execute流程与addWorker源码，对比Runnable/Callable、execute/submit、shutdown/shutdownNow，剖析Fixed、Single、Cached、Scheduled四种内置池的缺陷，给出ThreadPoolExecutor手动创建及阻塞队列选型最佳实践。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: Java 线程池详解category: Javatag:Java并发池化技术想必大家已经屡见不鲜了，线程池、数据库连接池、HTTP 连接池等等都是对这个思想的应用。池化技术的思想主要是为了减少每次获取资源的消耗，提高对资源的利用率。这篇文章我会详细介绍一下线程池的基本概念以及核心原理。线程池介绍顾名思义，线程池就是管理一系列线程的资源池，其提供了一种限制和管理线程资源的方式。每个线程池还维护一些基本统计信息，例如已完成任务的数量。这里借用《Java 并发编程的艺术》书中的部分内容来总结一下使用线程池的好处：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。线程池一般用于执行多个不相关联的耗时任务，没有多线程的情况下，任务顺序执行，使用了线程池的话可让多个不相关联的任务同时执行。Executor 框架介绍Executor 框架是 Java5 之后引进的，在 Java 5 之后，通过 Executor 来启动线程比使用 Thread 的 start 方法更好，除了更易管理，效率更好（用线程池实现，节约开销）外，还有关键的一点：有助于避免 this 逃逸问题。this 逃逸是指在构造函数返回之前其他线程就持有该对象的引用，调用尚未构造完全的对象的方法可能引发令人疑惑的错误。Executor 框架不仅包括了线程池的管理，还提供了线程工厂、队列以及拒绝策略等，Executor 框架让并发编程变得更加简单。Executor 框架结构主要由三大部分组成：1、任务(Runnable /Callable)执行任务需要实现的 Runnable 接口 或 Callable接口。Runnable 接口或 Callable 接口 实现类都可以被 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 执行。2、任务的执行(Executor)如下图所示，包括任务执行机制的核心接口 Executor ，以及继承自 Executor 接口的 ExecutorService 接口。ThreadPoolExecutor 和 ScheduledThreadPoolExecutor 这两个关键类实现了 ExecutorService 接口。这里提了很多底层的类关系，但是，实际上我们需要更多关注的是 ThreadPoolExecutor 这个类，这个类在我们实际使用线程池的过程中，使用频率还是非常高的。注意： 通过查看 ScheduledThreadPoolExecutor 源代码我们发现 ScheduledThreadPoolExecutor 实际上是继承了 ThreadPoolExecutor 并实现了 ScheduledExecutorService ，而 ScheduledExecutorService 又实现了 ExecutorService，正如我们上面给出的类关系图显示的一样。ThreadPoolExecutor 类描述://AbstractExecutorService实现了ExecutorService接口 public class ThreadPoolExecutor extends AbstractExecutorServiceScheduledThreadPoolExecutor 类描述://ScheduledExecutorService继承ExecutorService接口 public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService3、异步计算的结果(Future)Future 接口以及 Future 接口的实现类 FutureTask 类都可以代表异步计算的结果。当我们把 Runnable接口 或 Callable 接口 的实现类提交给 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 执行。（调用 submit() 方法时会返回一个 FutureTask 对象）Executor 框架的使用示意图：主线程首先要创建实现 Runnable 或者 Callable 接口的任务对象。把创建完成的实现 Runnable/Callable接口的 对象直接交给 ExecutorService 执行: ExecutorService.execute（Runnable command））或者也可以把 Runnable 对象或Callable 对象提交给 ExecutorService 执行（ExecutorService.submit（Runnable task）或 ExecutorService.submit（Callable <T> task））。如果执行 ExecutorService.submit（…），ExecutorService 将返回一个实现Future接口的对象（我们刚刚也提到过了执行 execute()方法和 submit()方法的区别，submit()会返回一个 FutureTask 对象）。由于 FutureTask 实现了 Runnable，我们也可以创建 FutureTask，然后直接交给 ExecutorService 执行。最后，主线程可以执行 FutureTask.get()方法来等待任务执行完成。主线程也可以执行 FutureTask.cancel（boolean mayInterruptIfRunning）来取消此任务的执行。ThreadPoolExecutor 类介绍（重要）线程池实现类 ThreadPoolExecutor 是 Executor 框架最核心的类。线程池参数分析ThreadPoolExecutor 类中提供的四个构造方法。我们来看最长的那个，其余三个都是在这个构造方法的基础上产生（其他几个构造方法说白点都是给定某些默认参数的构造方法比如默认制定拒绝策略是什么）。 /** * 用给定的初始参数创建一个新的ThreadPoolExecutor。 */ public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量 int maximumPoolSize,//线程池的最大线程数 long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间 TimeUnit unit,//时间单位 BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列 ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可 RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务 ) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }下面这些参数非常重要，在后面使用线程池的过程中你一定会用到！所以，务必拿着小本本记清楚。ThreadPoolExecutor 3 个最重要的参数：corePoolSize : 任务队列未达到队列容量时，最大可以同时运行的线程数量。maximumPoolSize : 任务队列中存放的任务达到队列容量的时候，当前可以同时运行的线程数量变为最大线程数。workQueue: 新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，新任务就会被存放在队列中。ThreadPoolExecutor其他常见参数 :keepAliveTime:线程池中的线程数量大于 corePoolSize 的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了 keepAliveTime才会被回收销毁。unit : keepAliveTime 参数的时间单位。threadFactory :executor 创建新线程的时候会用到。handler :饱和策略（后面会单独详细介绍一下）。下面这张图可以加深你对线程池中各个参数的相互关系的理解（图片来源：《Java 性能调优实战》）：ThreadPoolExecutor 饱和策略定义:如果当前同时运行的线程数量达到最大线程数量并且队列也已经被放满了任务时，ThreadPoolExecutor 定义一些策略:ThreadPoolExecutor.AbortPolicy：抛出 RejectedExecutionException来拒绝新任务的处理。ThreadPoolExecutor.CallerRunsPolicy：调用执行自己的线程运行任务，也就是直接在调用execute方法的线程中运行(run)被拒绝的任务，如果执行程序已关闭，则会丢弃该任务。因此这种策略会降低对于新任务提交速度，影响程序的整体性能。如果您的应用程序可以承受此延迟并且你要求任何一个任务请求都要被执行的话，你可以选择这个策略。ThreadPoolExecutor.DiscardPolicy：不处理新任务，直接丢弃掉。ThreadPoolExecutor.DiscardOldestPolicy：此策略将丢弃最早的未处理的任务请求。举个例子：举个例子：Spring 通过 ThreadPoolTaskExecutor 或者我们直接通过 ThreadPoolExecutor 的构造函数创建线程池的时候，当我们不指定 RejectedExecutionHandler 饱和策略来配置线程池的时候，默认使用的是 AbortPolicy。在这种饱和策略下，如果队列满了，ThreadPoolExecutor 将抛出 RejectedExecutionException 异常来拒绝新来的任务 ，这代表你将丢失对这个任务的处理。如果不想丢弃任务的话，可以使用CallerRunsPolicy。CallerRunsPolicy 和其他的几个策略不同，它既不会抛弃任务，也不会抛出异常，而是将任务回退给调用者，使用调用者的线程来执行任务public static class CallerRunsPolicy implements RejectedExecutionHandler { public CallerRunsPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { // 直接主线程执行，而不是线程池中的线程执行 r.run(); } } }线程池创建的两种方式方式一：通过ThreadPoolExecutor构造函数来创建（推荐）。方式二：通过 Executor 框架的工具类 Executors 来创建。Executors工具类提供的创建线程池的方法如下图所示：可以看出，通过Executors工具类可以创建多种类型的线程池，包括：FixedThreadPool：固定线程数量的线程池。该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务。SingleThreadExecutor： 只有一个线程的线程池。若多余一个任务被提交到该线程池，任务会被保存在一个任务队列中，待线程空闲，按先入先出的顺序执行队列中的任务。CachedThreadPool： 可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用。ScheduledThreadPool：给定的延迟后运行任务或者定期执行任务的线程池。《阿里巴巴 Java 开发手册》强制线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 构造函数的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险Executors 返回线程池对象的弊端如下(后文会详细介绍到)：FixedThreadPool 和 SingleThreadExecutor:使用的是无界的 LinkedBlockingQueue，任务队列最大长度为 Integer.MAX_VALUE,可能堆积大量的请求，从而导致 OOM。CachedThreadPool:使用的是同步队列 SynchronousQueue, 允许创建的线程数量为 Integer.MAX_VALUE ，如果任务数量过多且执行速度较慢，可能会创建大量的线程，从而导致 OOM。ScheduledThreadPool 和 SingleThreadScheduledExecutor:使用的无界的延迟阻塞队列DelayedWorkQueue，任务队列最大长度为 Integer.MAX_VALUE,可能堆积大量的请求，从而导致 OOM。// 无界队列 LinkedBlockingQueue public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()); } // 无界队列 LinkedBlockingQueue public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>())); } // 同步队列 SynchronousQueue，没有容量，最大线程数是 Integer.MAX_VALUE` public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); } // DelayedWorkQueue（延迟阻塞队列） public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }线程池常用的阻塞队列总结新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，新任务就会被存放在队列中。不同的线程池会选用不同的阻塞队列，我们可以结合内置线程池来分析。容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列）：FixedThreadPool 和 SingleThreadExector 。FixedThreadPool最多只能创建核心线程数的线程（核心线程数和最大线程数相等），SingleThreadExector只能创建一个线程（核心线程数和最大线程数都是 1），二者的任务队列永远不会被放满。SynchronousQueue（同步队列）：CachedThreadPool 。SynchronousQueue 没有容量，不存储元素，目的是保证对于提交的任务，如果有空闲线程，则使用空闲线程来处理；否则新建一个线程来处理任务。也就是说，CachedThreadPool 的最大线程数是 Integer.MAX_VALUE ，可以理解为线程数是可以无限扩展的，可能会创建大量线程，从而导致 OOM。DelayedWorkQueue（延迟阻塞队列）：ScheduledThreadPool 和 SingleThreadScheduledExecutor 。DelayedWorkQueue 的内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构，可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容原来容量的 1/2，即永远不会阻塞，最大扩容可达 Integer.MAX_VALUE，所以最多只能创建核心线程数的线程。线程池原理分析（重要）我们上面讲解了 Executor框架以及 ThreadPoolExecutor 类，下面让我们实战一下，来通过写一个 ThreadPoolExecutor 的小 Demo 来回顾上面的内容。线程池示例代码首先创建一个 Runnable 接口的实现类（当然也可以是 Callable 接口，我们后面会介绍两者的区别。）MyRunnable.javaimport java.util.Date; /** * 这是一个简单的Runnable类，需要大约5秒钟来执行其任务。 * @author shuang.kou */ public class MyRunnable implements Runnable { private String command; public MyRunnable(String s) { this.command = s; } @Override public void run() { System.out.println(Thread.currentThread().getName() + " Start. Time = " + new Date()); processCommand(); System.out.println(Thread.currentThread().getName() + " End. Time = " + new Date()); } private void processCommand() { try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public String toString() { return this.command; } } 编写测试程序，我们这里以阿里巴巴推荐的使用 ThreadPoolExecutor 构造函数自定义参数的方式来创建线程池。ThreadPoolExecutorDemo.javaimport java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class ThreadPoolExecutorDemo { private static final int CORE_POOL_SIZE = 5; private static final int MAX_POOL_SIZE = 10; private static final int QUEUE_CAPACITY = 100; private static final Long KEEP_ALIVE_TIME = 1L; public static void main(String[] args) { //使用阿里巴巴推荐的创建线程池的方式 //通过ThreadPoolExecutor构造函数自定义参数创建 ThreadPoolExecutor executor = new ThreadPoolExecutor( CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue<>(QUEUE_CAPACITY), new ThreadPoolExecutor.CallerRunsPolicy()); for (int i = 0; i < 10; i++) { //创建WorkerThread对象（WorkerThread类实现了Runnable 接口） Runnable worker = new MyRunnable("" + i); //执行Runnable executor.execute(worker); } //终止线程池 executor.shutdown(); while (!executor.isTerminated()) { } System.out.println("Finished all threads"); } } 可以看到我们上面的代码指定了：corePoolSize: 核心线程数为 5。maximumPoolSize：最大线程数 10keepAliveTime : 等待时间为 1L。unit: 等待时间的单位为 TimeUnit.SECONDS。workQueue：任务队列为 ArrayBlockingQueue，并且容量为 100;handler:饱和策略为 CallerRunsPolicy。输出结构：pool-1-thread-3 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-5 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-2 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-1 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-4 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-3 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-4 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-5 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-2 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-5 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-4 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-3 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-2 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-4 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-5 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-3 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-2 End. Time = Sun Apr 12 11:14:47 CST 2020 Finished all threads // 任务全部执行完了才会跳出来，因为executor.isTerminated()判断为true了才会跳出while循环，当且仅当调用 shutdown() 方法后，并且所有提交的任务完成后返回为 true 线程池原理分析我们通过前面的代码输出结果可以看出：线程池首先会先执行 5 个任务，然后这些任务有任务被执行完的话，就会去拿新的任务执行。 大家可以先通过上面讲解的内容，分析一下到底是咋回事？（自己独立思考一会）现在，我们就分析上面的输出内容来简单分析一下线程池原理。为了搞懂线程池的原理，我们需要首先分析一下 execute方法。 在示例代码中，我们使用 executor.execute(worker)来提交一个任务到线程池中去。这个方法非常重要，下面我们来看看它的源码： // 存放线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount) private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static int workerCountOf(int c) { return c & CAPACITY; } //任务队列 private final BlockingQueue<Runnable> workQueue; public void execute(Runnable command) { // 如果任务为null，则抛出异常。 if (command == null) throw new NullPointerException(); // ctl 中保存的线程池当前的一些状态信息 int c = ctl.get(); // 下面会涉及到 3 步 操作 // 1.首先判断当前线程池中执行的任务数量是否小于 corePoolSize // 如果小于的话，通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 2.如果当前执行的任务数量大于等于 corePoolSize 的时候就会走到这里，表明创建新的线程失败。 // 通过 isRunning 方法判断线程池状态，线程池处于 RUNNING 状态并且队列可以加入任务，该任务才会被加入进去 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 再次获取线程池状态，如果线程池状态不是 RUNNING 状态就需要从任务队列中移除任务，并尝试判断线程是否全部执行完毕。同时执行拒绝策略。 if (!isRunning(recheck) && remove(command)) reject(command); // 如果当前工作线程数量为0，新创建一个线程并执行。 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //3. 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 // 传入 false 代表增加线程时判断当前线程数是否少于 maxPoolSize //如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。 else if (!addWorker(command, false)) reject(command); }这里简单分析一下整个流程（对整个逻辑进行了简化，方便理解）：如果当前运行的线程数小于核心线程数，那么就会新建一个线程来执行任务。如果当前运行的线程数等于或大于核心线程数，但是小于最大线程数，那么就把该任务放入到任务队列里等待执行。如果向任务队列投放任务失败（任务队列已经满了），但是当前运行的线程数是小于最大线程数的，就新建一个线程来执行任务。如果当前运行的线程数已经等同于最大线程数了，新建线程将会使当前运行的线程超出最大线程数，那么当前任务会被拒绝，饱和策略会调用RejectedExecutionHandler.rejectedExecution()方法。在 execute 方法中，多次调用 addWorker 方法。addWorker 这个方法主要用来创建新的工作线程，如果返回 true 说明创建和启动工作线程成功，否则的话返回的就是 false。 // 全局锁，并发操作必备 private final ReentrantLock mainLock = new ReentrantLock(); // 跟踪线程池的最大大小，只有在持有全局锁mainLock的前提下才能访问此集合 private int largestPoolSize; // 工作线程集合，存放线程池中所有的（活跃的）工作线程，只有在持有全局锁mainLock的前提下才能访问此集合 private final HashSet<Worker> workers = new HashSet<>(); //获取线程池状态 private static int runStateOf(int c) { return c & ~CAPACITY; } //判断线程池的状态是否为 Running private static boolean isRunning(int c) { return c < SHUTDOWN; } /** * 添加新的工作线程到线程池 * @param firstTask 要执行 * @param core参数为true的话表示使用线程池的基本大小，为false使用线程池最大大小 * @return 添加成功就返回true否则返回false */ private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { //这两句用来获取线程池的状态 int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { //获取线程池中工作的线程的数量 int wc = workerCountOf(c); // core参数为false的话表明队列也满了，线程池大小变为 maximumPoolSize if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //原子操作将workcount的数量加1 if (compareAndIncrementWorkerCount(c)) break retry; // 如果线程的状态改变了就再次执行上述操作 c = ctl.get(); if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } // 标记工作线程是否启动成功 boolean workerStarted = false; // 标记工作线程是否创建成功 boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 加锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //获取线程池状态 int rs = runStateOf(ctl.get()); //rs < SHUTDOWN 如果线程池状态依然为RUNNING,并且线程的状态是存活的话，就会将工作线程添加到工作线程集合中 //(rs=SHUTDOWN && firstTask == null)如果线程池状态小于STOP，也就是RUNNING或者SHUTDOWN状态下，同时传入的任务实例firstTask为null，则需要添加到工作线程集合和启动新的Worker // firstTask == null证明只新建线程而不执行任务 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); //更新当前工作线程的最大容量 int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; // 工作线程是否启动成功 workerAdded = true; } } finally { // 释放锁 mainLock.unlock(); } //// 如果成功添加工作线程，则调用Worker内部的线程实例t的Thread#start()方法启动真实的线程实例 if (workerAdded) { t.start(); /// 标记线程启动成功 workerStarted = true; } } } finally { // 线程启动失败，需要从工作线程中移除对应的Worker if (! workerStarted) addWorkerFailed(w); } return workerStarted; }更多关于线程池源码分析的内容推荐这篇文章：硬核干货：4W 字从源码上分析 JUC 线程池 ThreadPoolExecutor 的实现原理现在，让我们在回到示例代码， 现在应该是不是很容易就可以搞懂它的原理了呢？没搞懂的话，也没关系，可以看看我的分析：我们在代码中模拟了 10 个任务，我们配置的核心线程数为 5、等待队列容量为 100 ，所以每次只可能存在 5 个任务同时执行，剩下的 5 个任务会被放到等待队列中去。当前的 5 个任务中如果有任务被执行完了，线程池就会去拿新的任务执行。几个常见的对比Runnable vs CallableRunnable自 Java 1.0 以来一直存在，但Callable仅在 Java 1.5 中引入,目的就是为了来处理Runnable不支持的用例。Runnable 接口不会返回结果或抛出检查异常，但是 Callable 接口可以。所以，如果任务不需要返回结果或抛出异常推荐使用 Runnable 接口，这样代码看起来会更加简洁。工具类 Executors 可以实现将 Runnable 对象转换成 Callable 对象。（Executors.callable(Runnable task) 或 Executors.callable(Runnable task, Object result)）。Runnable.java@FunctionalInterface public interface Runnable { /** * 被线程执行，没有返回值也无法抛出异常 */ public abstract void run(); }Callable.java@FunctionalInterface public interface Callable<V> { /** * 计算结果，或在无法这样做时抛出异常。 * @return 计算得出的结果 * @throws 如果无法计算结果，则抛出异常 */ V call() throws Exception; } execute() vs submit()execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；submit()方法用于提交需要返回值的任务。线程池会返回一个 Future 类型的对象，通过这个 Future 对象可以判断任务是否执行成功，并且可以通过 Future 的 get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用 get（long timeout，TimeUnit unit）方法的话，如果在 timeout 时间内任务还没有执行完，就会抛出 java.util.concurrent.TimeoutException。这里只是为了演示使用，推荐使用 ThreadPoolExecutor 构造方法来创建线程池。示例 1：使用 get()方法获取返回值。ExecutorService executorService = Executors.newFixedThreadPool(3); Future<String> submit = executorService.submit(() -> { try { Thread.sleep(5000L); } catch (InterruptedException e) { e.printStackTrace(); } return "abc"; }); String s = submit.get(); System.out.println(s); executorService.shutdown();输出：abc示例 2：使用 get（long timeout，TimeUnit unit）方法获取返回值。ExecutorService executorService = Executors.newFixedThreadPool(3); Future<String> submit = executorService.submit(() -> { try { Thread.sleep(5000L); } catch (InterruptedException e) { e.printStackTrace(); } return "abc"; }); String s = submit.get(3, TimeUnit.SECONDS); System.out.println(s); executorService.shutdown();输出：Exception in thread "main" java.util.concurrent.TimeoutException at java.util.concurrent.FutureTask.get(FutureTask.java:205)shutdown()VSshutdownNow()shutdown（） :关闭线程池，线程池的状态变为 SHUTDOWN。线程池不再接受新任务了，但是队列里的任务得执行完毕。shutdownNow（） :关闭线程池，线程池的状态变为 STOP。线程池会终止当前正在运行的任务，并停止处理排队的任务并返回正在等待执行的 List。isTerminated() VS isShutdown()isShutDown 当调用 shutdown() 方法后返回为 true。isTerminated 当调用 shutdown() 方法后，并且所有提交的任务完成后返回为 true几种常见的内置线程池FixedThreadPool介绍FixedThreadPool 被称为可重用固定线程数的线程池。通过 Executors 类中的相关源代码来看一下相关实现： /** * 创建一个可重用固定数量线程的线程池 */ public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory); }另外还有一个 FixedThreadPool 的实现方法，和上面的类似，所以这里不多做阐述： public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }从上面源代码可以看出新创建的 FixedThreadPool 的 corePoolSize 和 maximumPoolSize 都被设置为 nThreads，这个 nThreads 参数是我们使用的时候自己传递的。即使 maximumPoolSize 的值比 corePoolSize 大，也至多只会创建 corePoolSize 个线程。这是因为FixedThreadPool 使用的是容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列），队列永远不会被放满。执行任务过程介绍FixedThreadPool 的 execute() 方法运行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明：如果当前运行的线程数小于 corePoolSize， 如果再来新任务的话，就创建新的线程来执行任务；当前运行的线程数等于 corePoolSize 后， 如果再来新任务的话，会将任务加入 LinkedBlockingQueue；线程池中的线程执行完 手头的任务后，会在循环中反复从 LinkedBlockingQueue 中获取任务来执行；为什么不推荐使用FixedThreadPool？FixedThreadPool 使用无界队列 LinkedBlockingQueue（队列的容量为 Integer.MAX_VALUE）作为线程池的工作队列会对线程池带来如下影响：当线程池中的线程数达到 corePoolSize 后，新任务将在无界队列中等待，因此线程池中的线程数不会超过 corePoolSize；由于使用无界队列时 maximumPoolSize 将是一个无效参数，因为不可能存在任务队列满的情况。所以，通过创建 FixedThreadPool的源码可以看出创建的 FixedThreadPool 的 corePoolSize 和 maximumPoolSize 被设置为同一个值。由于 1 和 2，使用无界队列时 keepAliveTime 将是一个无效参数；运行中的 FixedThreadPool（未执行 shutdown()或 shutdownNow()）不会拒绝任务，在任务比较多的时候会导致 OOM（内存溢出）。SingleThreadExecutor介绍SingleThreadExecutor 是只有一个线程的线程池。下面看看SingleThreadExecutor 的实现： /** *返回只有一个线程的线程池 */ public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory)); } public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }从上面源代码可以看出新创建的 SingleThreadExecutor 的 corePoolSize 和 maximumPoolSize 都被设置为 1，其他参数和 FixedThreadPool 相同。执行任务过程介绍SingleThreadExecutor 的运行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明 :如果当前运行的线程数少于 corePoolSize，则创建一个新的线程执行任务；当前线程池中有一个运行的线程后，将任务加入 LinkedBlockingQueue线程执行完当前的任务后，会在循环中反复从LinkedBlockingQueue 中获取任务来执行；为什么不推荐使用SingleThreadExecutor？SingleThreadExecutor 和 FixedThreadPool 一样，使用的都是容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列）作为线程池的工作队列。SingleThreadExecutor 使用无界队列作为线程池的工作队列会对线程池带来的影响与 FixedThreadPool 相同。说简单点，就是可能会导致 OOM。CachedThreadPool介绍CachedThreadPool 是一个会根据需要创建新线程的线程池。下面通过源码来看看 CachedThreadPool 的实现： /** * 创建一个线程池，根据需要创建新线程，但会在先前构建的线程可用时重用它。 */ public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), threadFactory); } public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }CachedThreadPool 的corePoolSize 被设置为空（0），maximumPoolSize被设置为 Integer.MAX.VALUE，即它是无界的，这也就意味着如果主线程提交任务的速度高于 maximumPool 中线程处理任务的速度时，CachedThreadPool 会不断创建新的线程。极端情况下，这样会导致耗尽 cpu 和内存资源。执行任务过程介绍CachedThreadPool 的 execute() 方法的执行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明：首先执行 SynchronousQueue.offer(Runnable task) 提交任务到任务队列。如果当前 maximumPool 中有闲线程正在执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)，那么主线程执行 offer 操作与空闲线程执行的 poll 操作配对成功，主线程把任务交给空闲线程执行，execute()方法执行完成，否则执行下面的步骤 2；当初始 maximumPool 为空，或者 maximumPool 中没有空闲线程时，将没有线程执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)。这种情况下，步骤 1 将失败，此时 CachedThreadPool 会创建新线程执行任务，execute 方法执行完成；为什么不推荐使用CachedThreadPool？CachedThreadPool 使用的是同步队列 SynchronousQueue, 允许创建的线程数量为 Integer.MAX_VALUE ，可能会创建大量线程，从而导致 OOM。ScheduledThreadPool介绍ScheduledThreadPool 用来在给定的延迟后运行任务或者定期执行任务。这个在实际项目中基本不会被用到，也不推荐使用，大家只需要简单了解一下即可。public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }ScheduledThreadPool 是通过 ScheduledThreadPoolExecutor 创建的，使用的DelayedWorkQueue（延迟阻塞队列）作为线程池的任务队列。DelayedWorkQueue 的内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构，可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容原来容量的 1/2，即永远不会阻塞，最大扩容可达 Integer.MAX_VALUE，所以最多只能创建核心线程数的线程。ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor，所以创建 ScheduledThreadExecutor 本质也是创建一个 ThreadPoolExecutor 线程池，只是传入的参数不相同。public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorServiceScheduledThreadPoolExecutor 和 Timer 对比Timer 对系统时钟的变化敏感，ScheduledThreadPoolExecutor不是；Timer 只有一个执行线程，因此长时间运行的任务可以延迟其他任务。 ScheduledThreadPoolExecutor 可以配置任意数量的线程。 此外，如果你想（通过提供 ThreadFactory），你可以完全控制创建的线程;在TimerTask 中抛出的运行时异常会杀死一个线程，从而导致 Timer 死机即计划任务将不再运行。ScheduledThreadExecutor 不仅捕获运行时异常，还允许您在需要时处理它们（通过重写 afterExecute 方法ThreadPoolExecutor）。抛出异常的任务将被取消，但其他任务将继续运行。关于定时任务的详细介绍，可以看这篇文章：Java 定时任务详解 。线程池最佳实践Java 线程池最佳实践这篇文章总结了一些使用线程池的时候应该注意的东西，实际项目使用线程池之前可以看看。