Java進階專題(十五) 從電商系統角度研究多執行緒(下)
阿新 • • 發佈：2020-10-20
## 前言
本章節繼上章節繼續梳理：執行緒相關的基礎理論和工具、多執行緒程式下的效能調優和電商場景下多執行緒的使用。
## 多執行緒J·U·C
### ThreadLocal
#### 概念
ThreadLocal類並不是用來解決多執行緒環境下的共享變數問題，而是用來提供執行緒內部的共享變數。在多執行緒環境下，可以保證各個執行緒之間的變數互相隔離、相互獨立。
#### 使用
ThreadLocal例項一般定義為private static型別的，在一個執行緒內，該變數共享一份，類似上下文作用，可以用來上下傳遞資訊。
```java
public class ThreadLocalDemo implements Runnable{
private static ThreadLocal threadLocal = new ThreadLocal<>();
public void run(){
for (int i = 0; i < 3; i++) {
threadLocal.set(i);
System.out.println(Thread.currentThread().getName()+",value="+threadLocal.get()
);
}
}
public static void main(String[] args) {
ThreadLocalDemo demo = new ThreadLocalDemo();
new Thread(demo).start();
new Thread(demo).start();
}
}
```
結果分析：
同一個demo例項，不同的thread巢狀
結果列印了各自的變數值，執行緒內上下文被傳遞，不同執行緒間被隔離
#### 應用場景
資料庫連線，session管理
下面的基於日誌平臺的訪問鏈路追蹤中，會用到
#### 使用中遇到的坑
參與過一個專案，電商商鋪詳情頁凌晨排程生成。需要上下傳遞shopid，為每個商鋪重新生成一下。在商鋪詳情頁裡因為是按麵包屑分片生成，比如商鋪資訊、熱賣商品、最多好評、店主推薦、最新上架等。
其他資訊全部生成ok，唯獨商品列表多個列表出現問題。經查，在商品部分的查詢中用到了ThreadLocal，造成當前商鋪id丟失。
#### 原始碼解析
ThreadLocalMap是ThreadLocal內部類，由ThreadLocal建立，每個Thread裡維護一個
ThreadLocal. ThreadLocalMap型別的屬性threadLocals。所有的value值其實是儲存在
ThreadLocalMap中。
這個儲存結構的思路是反轉的...
**set方法**
```java
public void set(T value) {
//取到當前執行緒
Thread t = Thread.currentThread();
//從當前執行緒中拿出Map
ThreadLocalMap map = getMap(t);
if (map != null)
//如果非空，說明之前建立過了
//以當前建立的ThreadLocal物件為key，需要儲存的值為value，寫入Map
//因為每個執行緒Thread裡有自己獨自的Map，所以起到了隔離作用
map.set(this, value);
else
//如果沒有，那就建立
createMap(t, value);
}
```
**get方法**
```java
public T get() {
Thread t = Thread.currentThread();
//獲取到當前執行緒下的Map
ThreadLocalMap map = getMap(t);
if (map != null) {
//如果非空，根據當前ThreadLocal為key，取出對應的value即可
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
//如果map是空的，往往返回一個初始值，這是一個protect方法
//這就是為什麼建立ThreadLocal的時候往往要求實現這個方法
return setInitialValue();
}
```
**remove方法**
```java
public void remove() {
ThreadLocalMap m = getMap(Thread.currentThread());
//很簡單，獲取到map後，呼叫remove移除掉
if (m != null)
m.remove(this);
}
```
**ThreadLocal是如何避免記憶體洩漏的**
在上述的get方法中，Entry類繼承了WeakReference，即每個Entry物件都有一個ThreadLocal的弱引用，GC對於弱引用的物件採取積極的記憶體回收策略，避免無人搭理時發生記憶體洩露。
### Fork/Join
####概念
ForkJoin是由JDK1.7後提供多線併發處理框架。ForkJoinPool由Java大師Doug Lea主持編寫，處理邏輯大概分為兩步。
1.任務分割：Fork（分岔），先把大的任務分割成足夠小的子任務，如果子任務比較大的話還要對子任務進行繼續分割。
2.合併結果：join，分割後的子任務被多個執行緒執行後，再合併結果，得到最終的完整輸出。
#### 組成
- ForkJoinT ask：主要提供fork和join兩個方法用於任務拆分與合併；多數使用。RecursiveAction（無返回值的任務）和RecursiveT ask（需要返回值）來實現compute方法。
- ForkJoinPool：排程ForkJoinT ask的執行緒池；
- ForkJoinWorkerThread：Thread的子類，存放於執行緒池中的工作執行緒（Worker）；
- WorkQueue：任務佇列，用於儲存任務；
#### 基本使用
一個典型的例子：計算1-1000的和
```java
public class SumTask {
private static final Integer MAX = 100;
static class SubTask extends RecursiveTask {
// 子任務開始計算的值
private Integer start;
// 子任務結束計算的值
private Integer end;
public SubTask(Integer start , Integer end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if(end - start < MAX) {
//小於邊界，開始計算
System.out.println("start = " + start + ";end = " + end);
Integer totalValue = 0;
for(int index = this.start ; index <= this.end ; index++) {
totalValue += index;
}
return totalValue;
}else {
//否則，中間劈開繼續拆分
SubTask subTask1 = new SubTask(start, (start + end) / 2);
subTask1.fork();
SubTask subTask2 = new SubTask((start + end) / 2 + 1 , end);
subTask2.fork();
return subTask1.join() + subTask2.join();
}
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
Future taskFuture = pool.submit(new SubTask(1,1000));
try {
Integer result = taskFuture.get();
System.out.println("result = " + result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace(System.out);
}
}
}
```
#### 設計思想
- 普通執行緒池內部有兩個重要集合：工作執行緒集合，和任務佇列。
- ForkJoinPool也類似，工作集合裡放的是特殊執行緒ForkJoinWorkerThread，任務佇列裡放的是特殊任務ForkJoinTask
- 不同之處在於，普通執行緒池只有一個佇列。而ForkJoinPool的工作執行緒ForkJoinWorkerThread每個執行緒內都繫結一個雙端佇列。
- 在fork的時候，也就是任務拆分，將拆分的task會被當前執行緒放到自己的佇列中。
- 佇列中的任務被執行緒執行時，有兩種模式，預設是同步模式（asyncMode==false）從隊尾取任務（LIFO）
- 竊取：當自己佇列中執行完後，工作執行緒會到其他佇列的隊首獲取任務（FIFO），取到後如果任務再次fork，拆分會被放入當前執行緒的佇列，依次擴張
#### 注意
使用ForkJoin將相同的計算任務通過多執行緒執行。但是在使用中需要注意：
注意任務切分的粒度，也就是fork的界限。並非越小越好
判斷要不要使用ForkJoin。任務量不是太大的話，序列可能優於並行。因為多執行緒會涉及到上下文的切換
### Volatile
#### 概念
回顧Java 記憶體模型中的可見性、原子性和有序性：
1. 可見性，是指執行緒之間的可見性，一個執行緒修改的狀態對另一個執行緒是可見的
2. 原子性，指的是這個操作是原子不可拆分的，不允許別的執行緒中間插隊操作
3. 有序性指的是你寫的程式碼的順序要和最終執行的指令保持一致。因為在Java記憶體模型中，允許編譯器和處理器對指令進行重排序，重排序過程不會影響到單執行緒程式的執行，卻會影響到多執行緒併發執行的正確性。
volatile要解決的就是可見性和有序性問題。
#### 原理
Java記憶體模型分為主記憶體和執行緒工作記憶體兩大類。
主記憶體：多個執行緒共享的記憶體。方法區和堆屬於主記憶體區域。
執行緒工作記憶體：每個執行緒獨享的記憶體。虛擬機器棧、本地方法棧、程式計數器屬於執行緒獨享的工作記憶體
Java記憶體模型規定，所有變數都需要儲存在主記憶體中，執行緒需要時，在自己的工作記憶體儲存變數的副本，執行緒對變數的所有操作都在工作記憶體中進行，執行結束後再同步到主記憶體中去。這裡必然會存在時間差，在這個時間差內，該執行緒對副本的操作，對於其他執行緒是不見的，從而造成了可見性問題。
但是，當對volatile變數進行寫操作的時候，JVM會向處理器傳送一條lock字首的指令，將這個快取中的變量回寫到系統主存中。
同時，在多處理器下，為了保證各個處理器的快取是一致的，就會實現快取一致性協議。每個處理器通過嗅探在總線上傳播的資料來檢查自己快取的值是不是過期，一旦發現過期就會將當前處理器的快取行設定成無效狀態，強制從主記憶體讀取，這就保障了可見性。
而volatile變數，通過記憶體屏障可以禁止指令重排。從而實現指令的有序性。
#### 注意
volatile不能保證鎖的原子性。
案例：給前面的計數器案例里加上volatile試試
```java
public class BadVolatile {
private static volatile int i=0;
public int get(){
return i;
}
public void inc(){
int j=get();
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
i=j+1;
}
public static void main(String[] args) throws InterruptedException {
final BadVolatile counter = new BadVolatile();
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
public void run() {
counter.inc();
}
}).start();
}
Thread.sleep(3000);
//理論上10才對。可是....
System.out.println(counter.i);
}
}
```
達不到目的。說明原子性無法保障。
###ConcurrentHashMap
#### 基本使用
```java
public static void main(String[] args) throws InterruptedException {
//定義ConcurrentHashMap
Map map = new ConcurrentHashMap();
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
@Override
public void run() {
//多執行緒下的put可以放心使用
map.put(UUID.randomUUID().toString(), "1");
}
}).start();
}
Thread.sleep(3000);
System.out.println(map);
}
```
#### 原理
jdk1.7是分段鎖,1.8使用的cas+sychronized操作,具體看程式碼
```java
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//計算hash
int binCount = 0;
for (Node[] tab = table;;) {//自旋，確保插入成功
Node f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)
tab = initTable();//表為空的話，初始化表
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//否則，插入元素，看下面的 casTabAt 方法
//cas 在這裡！
if (casTabAt(tab, i, null, new Node(hash, key, value)))
break;
}
//...
else {
V oldVal = null;
//其他情況下，加鎖保持
//synchronized 在這裡！
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node pred = e;
if ((e = e.next) == null) {
pred.next = new Node(hash, key, value);
break;
}
}
}
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
if ((p = ((TreeBin)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
//compareAndSetObject，比較並插入，典型CAS操作
static final boolean casTabAt(Node[] tab, int i,
Node c, Node v) {
return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
//get取值
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
int h = spread(key.hashCode());
//判斷table是不是空的，當前桶上是不是空的
//如果為空，返回null
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//找到對應hash槽的第一個node，如果key相等，返回value
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//如果正在擴容，不影響，繼續順著node找即可
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//其他情況，逐個便利，比對key，找到後返回value
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
```
####總結：
put過程：
1.根據key的hash值定位到桶位置
2.如果table為空if，先初始化table。
3.如果table當前桶裡沒有node，cas新增元素。成功則跳出迴圈，失敗則進入下一輪for迴圈。
4.判斷是否有其他執行緒在擴容，有則幫忙擴容，擴容完成再新增元素。
5.如果桶的位置不為空，遍歷該桶的連結串列或者紅黑樹，若key已存在，則覆蓋，不存在則將key插入到連結串列或紅黑樹的尾部。
get過程：
1.根據key的hash值定位到桶位置。
2.map是否初始化，沒有初始化則返回null
3.定位的桶是否有頭結點，沒有返回null
4.是否有其他執行緒在擴容，有的話呼叫find方法沿node指標往後查詢。擴容與find可以並行，因為node的next指標不會變
5.若沒有其他執行緒在擴容，則遍歷桶對應的連結串列或者紅黑樹，使用equals方法進行比較。key相同則返回value,不存在則返回null
### 併發容器
1.ConcurrentHashMap
對應：HashMap
目標：代替Hashtable、synchronizedMap，使用最多，前面詳細介紹過
原理：JDK7中採用Segment分段鎖，JDK8中採用CAS+synchronized
2.CopyOnWriteArrayList
對應：ArrayList
目標：代替Vector、synchronizedList
原理：高併發往往是讀多寫少的特性，讀操作不加鎖，而對寫操作加Lock獨享鎖，先複製一份新的集
合，在新的集合上面修改，然後將新集合賦值給舊的引用，並通過volatile 保證其可見性。
檢視原始碼：volatile array，lock加鎖，陣列複製
3.CopyOnWriteArraySet
對應：HashSet
目標：代替synchronizedSet
原理：與CopyOnWriteArrayList實現原理類似。
4.ConcurrentSkipListMap
對應：TreeMap
目標：代替synchronizedSortedMap(TreeMap)
原理：基於Skip list（跳錶）來代替平衡樹，按照分層key上下連結指標來實現。
5.ConcurrentSkipListSet
對應：TreeSet
目標：代替synchronizedSortedSet(TreeSet)
原理：內部基於ConcurrentSkipListMap實現，原理一致
6.ConcurrentLinkedQueue
對應：LinkedList
對應：無界執行緒安全佇列
原理：通過隊首隊尾指標，以及Node類元素的next實現FIFO佇列
7.BlockingQueue
對應：Queue
特點：拓展了Queue，增加了可阻塞的插入和獲取等操作
原理：通過ReentrantLock實現執行緒安全，通過Condition實現阻塞和喚醒
實現類：
LinkedBlockingQueue：基於連結串列實現的可阻塞的FIFO佇列
ArrayBlockingQueue：基於陣列實現的可阻塞的FIFO佇列
PriorityBlockingQueue：按優先順序排序的佇列
##效能調優
### 鎖優化
#### Synchronized
使用synchronized時注意鎖粒度
```java
public synchronized void test(){
// TODO
}
public void test(){
synchronized (this) {
// TODO
}
}
```
```java
public static synchronized void test(){
// TODO
}
public static void test(){
synchronized (TestSynchronized.class) {
// TODO
}
}
```
#### Lock鎖優化
舉個例子：電商系統中記錄首頁被使用者瀏覽的次數，以及最後一次操作的時間（含讀或寫）。
```java
public class TotalLock {
//類建立的時間
final long start = System.currentTimeMillis();
//總耗時
AtomicLong totalTime = new AtomicLong(0);
//快取變數
private Map map = new HashMap(){{put("count",0L);}};
ReentrantLock lock = new ReentrantLock();
//檢視map被寫入了多少次
public Map read(){
lock.lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
//最後操作完成的時間
map.put("time",end);
lock.unlock();
System.out.println(Thread.currentThread().getName()+",read="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
//寫入
public Map write(){
lock.lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//寫入計數
map.put("count",map.get("count")+1);
long end = System.currentTimeMillis();
map.put("time",end);
lock.unlock();
System.out.println(Thread.currentThread().getName()+",write="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
public static void main(String[] args) throws InterruptedException {
TotalLock count = new TotalLock();
//讀
for (int i = 0; i < 4; i++) {
new Thread(()->{
count.read();
}).start();
}
//寫
for (int i = 0; i < 1; i++) {
new Thread(()->{
count.write();
}).start();
}
Thread.sleep(3000);
System.out.println(count.map);
System.out.println("讀寫總共耗時："+count.totalTime.get());
}
}
```
檢視後,我們發現檢視次數這裡其實是可以並行讀取的，我們關注的業務是寫入次數，也就是count，至於讀取發生的時間time的寫入操作，只是一個put，不需要原子性保障，對這個加互斥鎖沒有必要。改成讀寫鎖試試……
```java
public class ReadAndWrite {
//類建立的時間
final long start = System.currentTimeMillis();
//總耗時
AtomicLong totalTime = new AtomicLong(0);
//快取變數，注意！因為read併發，這裡換成ConcurrentHashMap
private Map map = new ConcurrentHashMap(){{put("count",0L);}};
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
//檢視map被寫入了多少次
public Map read(){
lock.readLock().lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
//最後操作完成的時間
map.put("time",end);
lock.readLock().unlock();
System.out.println(Thread.currentThread().getName()+",read="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
//寫入
public Map write(){
lock.writeLock().lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//寫入計數
map.put("count",map.get("count")+1);
long end = System.currentTimeMillis();
map.put("time",end);
lock.writeLock().unlock();
System.out.println(Thread.currentThread().getName()+",write="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
public static void main(String[] args) throws InterruptedException {
ReadAndWrite rw = new ReadAndWrite();
//讀
for (int i = 0; i < 4; i++) {
new Thread(()->{
rw.read();
}).start();
}
//寫
for (int i = 0; i < 1; i++) {
new Thread(()->{
rw.write();
}).start();
}
Thread.sleep(3000);
System.out.println(rw.map);
System.out.println("讀寫總共耗時："+rw.totalTime.get());
}
}
```
再來看讀的時間變化和總執行時間。當read遠大於write時，這個差距會更明顯
#### CAS樂觀鎖優化
舉例,直接用synchronized
```java
public class NormalSync implements Runnable{
Long start = System.currentTimeMillis();
int i=0;
public synchronized void run() {
int j = i;
//實際業務中可能會有一堆的耗時操作，這裡等待100ms模擬
try {
//做一系列操作
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//業務結束後，增加計數
i = j+1;
System.out.println(Thread.currentThread().getId()+
" ok,time="+(System.currentTimeMillis() - start));
}
public static void main(String[] args) throws InterruptedException {
NormalSync test = new NormalSync();
new Thread(test).start();
new Thread(test).start();
Thread.currentThread().sleep(1000);
System.out.println("last value="+test.i);
}
}
//執行緒二最終耗時會在200ms+，總耗時300ms，原因是悲觀鎖卡在了read後的耗時操作上，但是保證了
//最終結果是2
```
使用cas
```java
public class CasSync implements Runnable{
long start = System.currentTimeMillis();
int i=0;
public void run() {
int j = i;
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//CAS處理，在這裡理解思想，實際中不推薦大家使用！
try {
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
Unsafe unsafe = (Unsafe) f.get(null);
long offset =
unsafe.objectFieldOffset(CasSync.class.getDeclaredField("i"));
while (!unsafe.compareAndSwapInt(this,offset,j,j+1)){
j = i;
}
} catch (Exception e) {
e.printStackTrace();
}
//實際開發中，要用atomic包，或者while+synchronized自旋
// synchronized (this){
// //注意這裡！
// while (j != i){
// j = i;
// }
// i = j+1;
// }
System.out.println(Thread.currentThread().getName()+
" ok,time="+(System.currentTimeMillis() - start));
}
public static void main(String[] args) throws InterruptedException {
CasSync test = new CasSync();
new Thread(test).start();
new Thread(test).start();
Thread.currentThread().sleep(1000);
System.out.println("last value="+test.i);
}
}
//執行緒一、二均在100ms+，總耗時200ms，最終結果還是2
```
#### 總結
減少鎖的時間 不需要同步執行的程式碼，能不放在同步快裡面執行就不要放在同步快內，可以讓鎖儘快釋放
減少鎖的粒度 將物理上的一個鎖，拆成邏輯上的多個鎖，增加並行度，從而降低鎖競爭，典型如分段鎖
鎖的粒度 拆鎖的粒度不能無限拆，最多可以將一個鎖拆為當前cup數量相等
減少加減鎖的次數 假如有一個迴圈，迴圈內的操作需要加鎖，我們應該把鎖放到迴圈外面，否則每次進出迴圈，都要加鎖
使用讀寫鎖 業務細分，讀操作加讀鎖，可以併發讀，寫操作使用寫鎖，只能單執行緒寫，參考計數器案例
善用volatile volatile的控制比synchronized更輕量化，在某些變數上可以加以運用，如單例模式中
### 執行緒池引數優化
#### 一些經驗
1）corePoolSize
核心執行緒數，一旦有任務進來，在core範圍內會立刻建立執行緒進入工作。所以這個值應該參考業務併發量在絕大多數時間內的併發情況。同時分析任務的特性。
高併發，執行時間短的，要儘可能小的執行緒數，如配置CPU個數+1，減少執行緒上下文的切換。因為它不怎麼佔時間，讓少量執行緒快跑幹活。
併發不高、任務執行時間長的要分開看：如果時間都花在了IO上，那就調大，如配置兩倍CPU個數+1。不能讓CPU閒下來，執行緒多了並行處理更快。如果時間都花在了運算上，運算的任務還很重，本身就很佔cpu，那儘量減少cpu，減少切換時間。參考第一條
如果高併發，執行時間還很長……
2）workQueue
任務佇列，用於傳輸和儲存等待執行任務的阻塞佇列。這個需要根據你的業務可接受的等待時間。是一個需要權衡時間還是空間的地方，如果你的機器cpu資源緊張，jvm記憶體夠大，同時任務又不是那麼緊迫，減少coresize，加大這裡。如果你的cpu不是問題，對記憶體比較敏感比較害怕記憶體溢位，同時任務又要求快點響應。那麼減少這裡。
3）maximumPoolSize
執行緒池最大數量，這個值和佇列要搭配使用，如果你採用了無界佇列，那很大程度上，這個引數沒有意義。同時要注意，佇列盛滿，同時達到max的時候，再來的任務可能會丟失（下面的handler會講）。
如果你的任務波動較大，同時對任務波峰來的時候，實時性要求比較高。也就是來的很突然並且都是著急的。那麼調小佇列，加大這裡。如果你的任務不那麼著急，可以慢慢做，那就扔佇列吧。
佇列與max是一個權衡。佇列空間換時間，多花記憶體少佔cpu，輕視任務緊迫度。max捨得cpu執行緒開銷，少佔記憶體，給任務最快的響應。
4）keepaliveTime
執行緒存活保持時間，超出該時間後，執行緒會從max下降到core，很明顯，這個決定了你養閒人所花的代價。如果你不缺cpu，同時任務來的時間沒法琢磨，波峰波谷的間隔比較短。經常性的來一波。那麼實當的延長銷燬時間，避免頻繁建立和銷燬執行緒帶來的開銷。如果你的任務波峰出現後，很長一段時間不再出現，間隔比較久，那麼要適當調小該值，讓閒著不幹活的執行緒儘快銷燬，不要佔據資源。
5）threadFactory（自定義展示例項）
執行緒工廠，用於建立新執行緒。threadFactory建立的執行緒也是採用new Thread()方式，threadFactory建立的執行緒名都具有統一的風格：pool-m-thread-n（m為執行緒池的編號，n為執行緒池內的執行緒編號）。如果需要自己定義執行緒的某些屬性，如個性化的執行緒名，可以在這裡動手。一般不需要折騰它。
6）handler
執行緒飽和策略，當執行緒池和佇列都滿了，再加入執行緒會執行此策略。預設不處理的話會扔出異常，打進日誌。這個與任務處理的資料重要程度有關。如果資料是可丟棄的，那不需要額外處理。如果資料極其重要，那需要在這裡採取措施防止資料丟失，如扔訊息佇列或者至少詳細打入日誌檔案可追蹤。
#### 併發容器選擇
案例一：電商網站中記錄一次活動下各個商品售賣的數量。
場景分析：需要頻繁按商品id做get和set，但是商品id（key）的數量相對穩定不會頻繁增刪
初級方案：選用HashMap，key為商品id，value為商品購買的次數。每次下單取出次數，增加後再寫入
問題：HashMap執行緒不安全！在多次商品id寫入後，如果發生擴容，在JDK1.7 之前，在併發場景下HashMap 會出現死迴圈，從而導致CPU 使用率居高不下。JDK1.8 中修復了HashMap 擴容導致的死迴圈問題，但在高併發場景下，依然會有資料丟失以及不準確的情況出現。
選型：Hashtable 不推薦，鎖太重，選ConcurrentHashMap 確保高併發下多執行緒的安全性
案例二：在一次活動下，為每個使用者記錄瀏覽商品的歷史和次數。
場景分析：每個使用者各自瀏覽的商品量級非常大，並且每次訪問都要更新次數，頻繁讀寫
初級方案：為確保執行緒安全，採用上面的思路，ConcurrentHashMap
問題：ConcurrentHashMap 內部機制在資料量大時，會把連結串列轉換為紅黑樹。而紅黑樹在高併發情況下，刪除和插入過程中有個平衡的過程，會牽涉到大量節點，因此競爭鎖資源的代價相對比較高
選型：用跳錶，ConcurrentSkipListMap將key值分層，逐個切段，增刪效率高於ConcurrentHashMap
結論：如果對資料有強一致要求，則需使用Hashtable；在大部分場景通常都是弱一致性的情況下，使用ConcurrentHashMap 即可；如果資料量級很高，且存在大量增刪改操作，則可以考慮使用
ConcurrentSkipListMap。
案例三：在活動中，建立一個使用者列表，記錄凍結的使用者。一旦凍結，不允許再下單搶購，但是可
以瀏覽。
場景分析：違規被凍結的使用者不會太多，但是絕大多數非凍結使用者每次搶單都要去查一下這個列表。低頻寫，高頻讀。
初級方案：ArrayList記錄要凍結的使用者id
問題：ArrayList對凍結使用者id的插入和讀取操作在高併發時，執行緒不安全。Vector可以做到執行緒安全，但併發效能差，鎖太重。
選型：綜合業務場景，選CopyOnWriteArrayList，會佔空間，但是也僅僅發生在新增新凍結使用者的時候。絕大多數的訪問在非凍結使用者的讀取和比對上，不會