1、池化技術（思想）
（1）池化技術：程式運行的本質->占用系統的資源，優化資源的使用便是池化技術
（2）簡單理解：實作準備好一些資源，有人要用，就來這個池拿，用完放回即可，執行緒池、資料庫連接池、Http連接池等待
（3）好處：主要是為了減少每次獲取資源的消耗，提高對資源的利用率，
2、執行緒池的好處
（1）降低資源消耗
通過重復利用已創建的執行緒降低執行緒創建和銷毀造成的消耗
（2）提高相應速度
當任務到達時，任務可以不需要等待執行緒創建就能立即執行
（3）提高執行緒的可管理性
執行緒是稀缺資源，無限制創建會消耗系統資源，還會降低系統的穩定性，使用執行緒池可以進行統一的分配，調優和監控，
3、Executor（一般創建執行緒池不建議使用）
（1）通過Executor來啟動執行緒與Thread.start()方法相比：更好
==用執行緒池實作，節約開銷，容易管理，效率更好
==有助于避免this逃逸問題
====this逃逸：指在建構式回傳之前其它執行緒就持有該物件的參考，呼叫未構造完全的物件的方法，可能引發錯誤，
（2）Executor框架，包括執行緒池的管理、執行緒工廠、佇列、拒絕策略等，并發編程更加簡單，
（3）結構：任務（Runnable\Callable）、任務的執行（Executor）、異步計算的結果（Future）(3大部分)
==【1】任務（Runnable\Callable）
====1)實作Runnable介面或Callable介面
====2)兩個介面的實作類都可以被ThreadPoolExecutor（重要）或ScheduledThreadPoolExecutor執行，
==【2】任務的執行（Executor）
====1)Executor介面：任務執行機制的核心介面
====2)ExecutorService介面：繼承自Executor介面
====3)ThreadPoolExecutor（重要）和ScheduledThreadPoolExecutor兩個關鍵類實作了ExecutorService介面
//AbstractExecutorService實作了ExecutorService介面
public class ThreadPoolExecutor extends AbstractExecutorService
====4)ScheduledThreadPoolExecutor實際上是繼承了ThreadPoolExecutor（重要）并實作了ScheduledExecutorSercice
//ScheduledExecutorService繼承ExecutorService介面
public class ScheduledThreadPoolExecutor
extends ThreadPoolExecutor
implements ScheduledExecutorService
==【3】異步計算的結果（Future）
====1)Future介面以及Future介面的實作類FutureTask類都可以代表異步計算的結果
====2)把Runnable介面或Callable介面的實作類提交給ThreadPoolExecutor或ScheduledThreadPoolExecutor執行，呼叫submit()方法時會回傳一個FutureTask物件，
（4）Executor的使用
==[1]主執行緒創建實作Runnable介面或Callable介面的任務物件
==[2]任務物件直接交給ExecutorService介面執行
====ExecutorService.execute(Runnable command)
或
====ExecutorService.submit(Runnable task)
或
====ExecutorService.submit(Callable<T> task)
==[3]獲取Future介面的物件——FutureTask（submit()和execute()的區別）
====如果執行submit(),ExecutorService會回傳一個FutureTask物件（submit用于提交需要回傳值的任務），通過FutureTask物件可以判斷是否執行成功，同時，可以通過這個物件的get()方法來獲取回傳值
======FutureTask物件的get()方法，會阻塞當前執行緒直到任務完成，也可使用get(long time,TimeUnit timeunit)方法，阻塞一段時間后立即回傳，但可能任務沒有執行完，會拋出java.util.concurrent.TimeOutException，
====如果 執行execute(),ExecutorService不會有回傳值（用于提交不需要回傳值的任務）,無法判斷任務是否被執行緒池執行成功與否，
====由于FutureTask實作了Runnable,可以創建FutureTask，直接交給ExecutorService執行，
==[4]主執行緒可以執行FutureTask.get()方法等待任務執行完成，也可以執行FutureTask.cancel(boolean mayInterruptIfRunning)方法取消任務的執行，
4、ThreadPoolExecutor類(重要)
（1）有四個構造方法（決議最長的一個）
/**
* 用給定的初始引數創建一個新的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;
}
（2）ThreadPoolExecutor七大引數（3個最重要）
==[1]corePoolSize:
==== 核心執行緒大小，核心執行緒數定義了最小可以同時運行的核心執行緒數量，
====執行緒池一直運行，核心執行緒就不會停止
==[2]maximumPoolSize:
====執行緒池最大執行緒數量
====非核心執行緒數量等于maximumPoolSize-corePoolSize
====當佇列中存放的任務達到佇列容量時，當前可以同時運行的執行緒數量變為最大執行緒數
==[3]workQueue:
====作業佇列，包含阻塞佇列,ArrayBlockingQueue,LinkedBlockingQueue等
====用來存放執行緒任務，當新任務來的時候，會先判斷當前運行的執行緒數量是否達到核心執行緒數，如果達到，新任務就會被存放在佇列中，
==[4]keepAliveTime：
====非核心執行緒的心跳時間，非核心執行緒在keepAliveTime時間內沒有運行任務，且等待時間超過keepAliveTime，就會被回收銷毀
====執行緒池中執行緒數量大于corePoolSize，超出的為非核心執行緒
==[5]unit:keepAliveTime引數的時間單位
==[6]threadFactory:
====執行緒工廠，新建執行緒的工廠
====創建新執行緒時會用到
==[7]handler：飽和(拒絕)策略（通過實作RejectedExecutionHandler介面）
====如果當前同時運行的執行緒數量達到最大執行緒數量且佇列也已放滿了任務，就要使用ThreadPoolExecutor定義的策略
====ThreadPoolExecutor.AbortPolicy（默認）
執行緒任務拒絕，拒絕處理新任務，拋出RejectedExecutionException錯誤
====ThreadPoolExecutor.CallerRunsPolicy
（執行緒池外的執行緒直接呼叫run()方法執行）
呼叫執行自己的執行緒運行任務，即直接在呼叫execute()方法的執行緒中，直接呼叫run()方法運行被拒絕的任務；
如果執行的程式已經關閉，則丟棄該任務
因此，這種策略會降低對于新任務的提交速度，影響程式的整體性能
====ThreadPoolExecutor.DiscardPolicy
新任務直接丟棄，不報錯，即不處理新任務
====ThreadPoolExecutor.DiscardOldestPolicy
將workQueue隊首的任務丟棄，將最新的任務重新加入佇列等執行
丟棄最早的未處理的任務(佇列隊首)的請求
5、實作執行緒池
（1）通過ThreadPoolExecutor建構式實作（推薦）
（2）通過Executor框架的工具類Executors實作（不建議）
6、ThreadPoolExecutor的使用
（1）自定義(ExecutorService)
//自定義 執行緒池
public class dMyThreadPoolTest {
public static void main(String[] args) {
ExecutorService myThreadPool = new ThreadPoolExecutor(
//最大承載數 = 最大執行緒池大小 + 阻塞佇列大小
2,//核心執行緒池大小
5,//最大執行緒池大小
2,//超時等待時間
TimeUnit.SECONDS,//超時等待時間單位
new LinkedBlockingDeque<>(3),//阻塞佇列為3 即有3個等待位置
Executors.defaultThreadFactory(),//默認的執行緒創建工廠
// new ThreadPoolExecutor.AbortPolicy()//第一種 拒絕策略 有超過最大承載數的執行緒時 不處理 會拋出例外 （銀行滿了，還有人進來不處理這個人 拋出例外） java.util.concurrent.RejectedExecutionException
// new ThreadPoolExecutor.CallerRunsPolicy()//第二種拒絕策略 超過承載數的 哪來的去哪里 即由哪里實作 這里是main執行緒來的 就由main執行緒實作
// new ThreadPoolExecutor.DiscardPolicy() //第三種拒絕策略 超過承載數的 就丟掉 不拋出例外 也不實作
new ThreadPoolExecutor.DiscardOldestPolicy()//第四種拒絕策略 超過承載數的 會嘗試與最早的執行緒爭取實作 爭取成功就實作 失敗就丟掉 不拋出例外
);
?
try {
for (int i = 1; i <= 9; i++) {
myThreadPool.execute(()->{
System.out.println(Thread.currentThread().getName()+" 666");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
myThreadPool.shutdown();
}
?
}
}
（2）CPU密集型寫法（最大執行緒池=CPU核數）(ExecutorService)
//CPU密集型寫法 最大執行緒池 定為 CPU的核數
public class eMyThreadPoolTestCPU {
public static void main(String[] args) {
//獲取CPU的核數
System.out.println(Runtime.getRuntime().availableProcessors());
ExecutorService myThreadPool = new ThreadPoolExecutor(
2,
Runtime.getRuntime().availableProcessors(), //最大執行緒池為CPU的核數
2,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(3),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardOldestPolicy()
);
?
try {
for (int i = 0; i < 8; i++) {
myThreadPool.execute(()->{
System.out.println(Thread.currentThread().getName()+" 666");
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
myThreadPool.shutdown();
}
}
}
(3)實作Runnable介面
import java.util.Date;
?
/**
* 這是一個簡單的Runnable類，需要大約5秒鐘來執行其任務，
*/
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構造
?
import 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 ：最大執行緒數 10
//keepAliveTime : 等待時間為 1L，
//unit: 等待時間的單位為 TimeUnit.SECONDS，
//workQueue：任務佇列為 ArrayBlockingQueue，并且容量為 100;
//handler:飽和策略為 CallerRunsPolicy，
7、執行緒池原理分析
（1）execute()\submit()方法：提交一個任務到執行緒池中
execute()底層：
addWorker():用來創建新的作業執行緒，回傳true即創建和啟動執行緒成功
（2）配置核心執行緒數量N、配置等待佇列容量M、模擬N+M+S個任務
====每次只能存在N個任務同時執行，剩下的M個任務會被放到等待佇列中,S個任務會放到執行緒池中
====當前N個任務有被執行完成的，佇列中M個任務一些放入執行，S執行緒池會去拿新的任務
（3）1、執行緒池執行execute()|submit()方法向執行緒池中添加任務，當任務小于核心執行緒數corePoolSize，執行緒池中可以創建新的執行緒
2、當任務大于核心執行緒池corePoolSize，就向佇列workQueue添加任務
3、當佇列workQueue已經滿了,需要比較maximumPoolSize，若小于maximumPoolSize,則在執行緒池創建新的執行緒，若大于maximumPoolSize，則說明當前設定執行緒池中執行緒已經處理不了了，就會執行飽和（拒絕）策略，
8、對比
（1）Runnable和Callable
==Runnable介面不會回傳結果或拋出檢查例外
@FunctionalInterface
public interface Runnable {
/**
* 被執行緒執行，沒有回傳值也無法拋出例外
*/
public abstract void run();
}
==Callable介面會回傳結果或拋出檢查例外
@FunctionalInterface
public interface Callable<V> {
/**
* 計算結果，或在無法這樣做時拋出例外，
* @return 計算得出的結果
* @throws 如果無法計算結果，則拋出例外
*/
V call() throws Exception;
}
（2）execute()和submit()
==execute()：用于提交不需要回傳值的任務，無法判斷任務是否被執行緒池執行成功與否
==submit()：用于提交需要回傳值的任務，回傳FutureTask物件
FutureTask物件可以判斷任務是否執行成功，并且可以通過get()方法獲取回傳值
get()方法會阻塞當前執行緒直到任務完成
get(Long timeout,TimeUnit unit)會阻塞，但在timeout時間內任務沒有執行完會拋出java.util.concurrent.TimeOutException
(3)shutdown()和shutdownNow()
==shutdown():關閉執行緒池，執行緒池的狀態變為SHUTDOWN,不再接受新的任務，但佇列里的任務得執行完成
==shutdownNow():關閉執行緒池，執行緒池的狀態變為STOP,執行緒池會終止當前正在運行的任務，并停止處理排隊的任務并回傳正在等待執行的List.
(4)Executor和Executors
==Executor是介面物件，能執行執行緒任務，ExecutorService介面繼承了Executor介面并進行了擴展，提供了獲得任務執行狀態和任務回傳值的方法；使用ThreadPoolExecutor可以創建自定義執行緒池，Future表示異步計算結果，提供了檢查計算是否完成的方法，以等待計算的完成，使用get()可以獲取計算的結果
==Executors是工具類，可使用其中不同的方法來實作我們的需求來創建執行緒池
9、常見的執行緒池
（1）FixedThreadPool:可重用的指定執行緒數的執行緒池
corePoolSize和maximumPoolSize是我們自己指定的
/**
* 創建一個可重用固定數量執行緒的執行緒池
corePoolSize和maximumPoolSize是我們自己指定的
*/
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
==執行程序：
====若當前運行的執行緒數小于corePoolSize，如果有新任務，就創建新的執行緒來執行任務
====若當前運行的執行緒等于corePoolSize，有新任務時，會將任務加入到LinkedBlockingQueue
====執行緒池中的執行緒執行完后，會回圈從LinkedBlockingQueue中獲取任務來執行
==不建議使用FixedThreadPool
====使用了無界的LinkedBlockingQueue(佇列容量為Integer.Max_Value，可認為是無界的)
====執行緒池中執行緒數達到corePoolSize后，新任務會在無界佇列中等待，因此執行緒池中的執行緒數不會超過corePoolSize
====不存在佇列滿的情況，可以設定corePoolSize和maximumPoolSize同值，maximumPoolSize無效
====無界佇列，keepAliveTime將無效
====運行中FixedThreadPool不會拒絕任務，任務多的時候可能會造成記憶體溢位（OOM）
（2）SingleThreadExecutor:只有一個執行緒的執行緒池
corePoolSize和maximumPoolSize被設定為1
/**
*回傳只有一個執行緒的執行緒池
corePoolSize和maximumPoolSize被設定為1
*/
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
==執行程序：
====只會用唯一的作業執行緒來執行任務，保證所有的任務按照指定的順序（FIFO，LIFO，優先級）執行
====若當前運行執行緒數少于corePoolSize，則創建一個新的執行緒執行任務
====當前執行緒池中有一個運行的執行緒時，將任務加入LinkedBlockingQueue
====執行緒執行完當前的任務后，會從LinkedBlockingQueue中獲取任務來執行
==特定：保證順序地執行每個任務，且在任意給定時間內不會有多個執行緒是活動的
==不建議使用SingleThreadExecutor
====使用無界佇列LinkedBlockingQueue作為作業佇列
====SingleThreadExecutor使用無界佇列,不會拒絕，會給執行緒池造成OOM(同FixedThreadPool)
（3）CachedThreadPool:可快取的，會根據需要創建新執行緒的執行緒池
corePoolSize為0 maximumPoolSize為Integer.Max_Value(無界)
/**
* 創建一個執行緒池，根據需要創建新執行緒，但會在先前構建的執行緒可用時重用它，
corePoolSize為0
maximumPoolSize為Integer.Max_Value(無界)
*/
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
}
==執行程序
====執行緒池長度超過處理需要，可靈活回收空閑執行緒，若無可回收，就新建執行緒
====首先執行SynchronousQueue.offer(Runnable task)提交任務到任務佇列
若當前maximunPool中有空閑執行緒正在執行，SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)，主執行緒執行offer操作與空閑執行緒執行poll()操作配對成功，主執行緒把任務交給空閑執行緒執行，execute()方法執行完成否則執行以下步驟
====當初始maximumPool為空，或maximumPoo中沒有空閑執行緒，將沒有執行緒執行SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)，此時，CachedThreadPool會創建新執行緒執行任務，execute方法執行完成，
==不建議使用
CachedThreadPool允許創建的執行緒數量為Integer.Max_Value，當主執行緒提交任務的速度高于maximumPool中的執行緒處理任務的速度，可能會不斷創建大量的執行緒，造成OOM
（4）ScheduledThreadPoolExecutor：定長的，支持定時的以及周期性任務執行
主要用來在給定的延遲后運行任務，或者定期執行任務
==使用任務佇列：DelayWorkQueue
====封裝了PriorityQueue,會對佇列中的任務進行排序，執行所需時間短的放在前面先執行（ScheduledFutureTask的time變數小的先執行），執行所需時間相同則先提交的任務先執行（ScheduledFutureTask的squenceNumber變數小的先執行）
====Timer對系統時鐘變化敏感，只有一個執行執行緒，因此，長時間運行的任務可以延遲其它任務
====ScheduledThreadPoolExecutor可以配置任意數量的執行緒，可以完全控制創建的執行緒
====TimerTask中拋出的運行時例外會kill一個執行緒，導致Timer死機
即當前拋出例外的任務不再允許，允許在需要處理（通過重寫afterExecute()），其它任務將繼續運行，
==執行程序：
====當呼叫SchedulledThreadPoolExecutor.scheduleAtFixedRate()方法或scheduleWithFixedDelay()方法，會向DelayQueue添加一個實作了RunnableScheduledFuture介面的ScheduledFutureTask，
====執行緒池中的執行緒從DelayQueue中獲取ScheduledFutureTask，然后執行任務
10、執行緒池大小的確定（如何合理配置執行緒池的引數）
背景關系切換： 當前任務在執行完 CPU 時間片切換到另一個任務之前會先保存自己的狀態，以便下次再切換回這個任務時，可以再加載這個任務的狀態，任務從保存到再加載的程序就是一次背景關系切換，
（1）CPU密集型
==該任務需要最大的運算，而沒有阻塞，CPU一直全速運行，
==任務消耗主要是CPU資源
==可將執行緒數設定為N(CPU核數)+1
== +1的目的：防止執行緒可能發生的缺頁中斷或導致任務暫停帶來的影響
（2）IO密集型
==該任務需要大量的IO,即大量的阻塞，系統大部分時間在處理IO互動
==處理IO時間段內不會占用太多的CPU，可以將CPU交給其它執行緒使用
==可將執行緒數設定為2N
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/458458.html
標籤：其他
上一篇：python中的函式
下一篇：C++基礎01-類與物件