為什么要用線(xiàn)程池線(xiàn)程是不是越多好？ 線(xiàn)程在Java中是一個(gè)對(duì)象， 更是操作系統(tǒng)的資源， 線(xiàn)程額創(chuàng)建和銷(xiāo)毀都需要時(shí)間，如果 創(chuàng)建時(shí)間 銷(xiāo)毀時(shí)間>執(zhí)行任務(wù)時(shí)間 就很不合算 Java對(duì)象占用堆內(nèi)存，操作系統(tǒng)線(xiàn)程占用系統(tǒng)內(nèi)存， 根據(jù)JVM規(guī)范，一個(gè)線(xiàn)程默認(rèn)最大棧大小為1M， 這個(gè)?？臻g是要從操作系統(tǒng)內(nèi)存中分配的，線(xiàn)程過(guò)多會(huì)消耗很多的內(nèi)存 操作系統(tǒng)頻繁切換線(xiàn)程上下文會(huì)影響性能
線(xiàn)程池的推出就是為了控制線(xiàn)程數(shù)量 線(xiàn)程池原理 - 概念線(xiàn)程池管理器：用于創(chuàng)建并管理線(xiàn)程池， 包括創(chuàng)建線(xiàn)程池， 銷(xiāo)毀線(xiàn)程池， 添加新任務(wù) 工作線(xiàn)程： 線(xiàn)程池中的線(xiàn)程， 在沒(méi)有任務(wù)時(shí)處于等待狀態(tài)， 可以循環(huán)的執(zhí)行任務(wù) 任務(wù)接口： 每個(gè)任務(wù)必須實(shí)現(xiàn)的接口，以供工作線(xiàn)程任務(wù)調(diào)度的執(zhí)行， 它主要規(guī)定了任務(wù)的入口，任務(wù)執(zhí)行完后的收尾工作，任務(wù)的執(zhí)行狀態(tài)等 任務(wù)隊(duì)列： 用于存放沒(méi)有處理的任務(wù)，提供一種緩沖
線(xiàn)程池API - 接口定義和實(shí)現(xiàn)類(lèi)
線(xiàn)程池API - 方法定義ExecutorService 
ScheduledExecutorService 
線(xiàn)程池API - Executors工具類(lèi)可以自己實(shí)例化線(xiàn)程池， 也可以用Executors創(chuàng)建線(xiàn)程池，方法如下： newFixedThreadPool(int nThreads) : 創(chuàng)建一個(gè)固定大小，任務(wù)隊(duì)列無(wú)界的的線(xiàn)程池，核心線(xiàn)程數(shù)=最大線(xiàn)程數(shù) newCachedThreadPool() : 創(chuàng)建一個(gè)大小無(wú)界的緩沖線(xiàn)程池，它的任務(wù)隊(duì)列是一個(gè)同步隊(duì)列，任務(wù)加入加入到池中，如果池中有空閑線(xiàn)程， 則用空閑線(xiàn)程執(zhí)行，如無(wú)則創(chuàng)建新線(xiàn)程執(zhí)行，池中的空閑線(xiàn)程超過(guò)60秒， 將被銷(xiāo)毀釋放， 線(xiàn)程數(shù)隨任務(wù)的多少變化，適用于耗時(shí)較小的異步任務(wù)， 池的核心線(xiàn)程數(shù)=0， 最大線(xiàn)程數(shù)=Integer.MAX_VALUE newSingleThreadExecutor() : 只有一個(gè)線(xiàn)程來(lái)執(zhí)行無(wú)界任務(wù)隊(duì)列的單一線(xiàn)程池，該線(xiàn)程池確保任務(wù)按加入的順序一個(gè)一個(gè)的依次執(zhí)行，當(dāng)唯一的線(xiàn)程因任務(wù)異常中止時(shí)，將創(chuàng)建一個(gè)新的線(xiàn)程來(lái)繼續(xù)執(zhí)行后續(xù)的任務(wù)， 與newFixedThreadPool(1)的區(qū)別在于，單一線(xiàn)程池的池大小在newSingleThreadExecutor方法中硬編碼，不能再改變 newScheduledThreadPool(int corePoolSize) : 能定時(shí)執(zhí)行任務(wù)的線(xiàn)程池，該池的核心線(xiàn)程數(shù)由參數(shù)指定，最大線(xiàn)程數(shù)=Integer.MAX_VALUE
線(xiàn)程池原理 - 任務(wù)執(zhí)行過(guò)程是否達(dá)到核心線(xiàn)程數(shù)量？ 沒(méi)達(dá)到，創(chuàng)建一個(gè)工作線(xiàn)程來(lái)執(zhí)行任務(wù) 工作隊(duì)列是否已滿(mǎn)？沒(méi)滿(mǎn)，則將新提交的任務(wù)存儲(chǔ)在任務(wù)隊(duì)列中 是否達(dá)到線(xiàn)程池最大數(shù)量？則創(chuàng)建一個(gè)新的線(xiàn)程來(lái)執(zhí)行任務(wù) 最后，執(zhí)行拒絕策略來(lái)處理這個(gè)任務(wù)
代碼示例： package com.neteasy.demo;
import java.util.List;
import java.util.concurrent.*;
/** 線(xiàn)程池的使用 */
public class Demo7 {
/**
* 測(cè)試： 提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù),看線(xiàn)程池的狀況
*
* @param threadPoolExecutor
* @throws Exception
*/
private void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
// 測(cè)試： 提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù)，看超過(guò)大小的2個(gè)，對(duì)應(yīng)的處理情況
for (int i = 0; i < 15; i ) {
int n = i;
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
try {
System.out.println("開(kāi)始執(zhí)行： " n);
Thread.sleep(3000L);
System.out.println("執(zhí)行結(jié)束： " n);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
System.out.println("任務(wù)提交成功：" i);
}
// 查看線(xiàn)程數(shù)量，查看隊(duì)列等待數(shù)量
Thread.sleep(500L);
System.out.println("當(dāng)前線(xiàn)程池線(xiàn)程數(shù)量為：" threadPoolExecutor.getPoolSize());
System.out.println("當(dāng)前線(xiàn)程池等待的數(shù)量為" threadPoolExecutor.getQueue().size());
// 等待15秒，查看線(xiàn)程數(shù)量和隊(duì)列數(shù)量（理論上，會(huì)被超出核心線(xiàn)程數(shù)量的線(xiàn)程自動(dòng)銷(xiāo)毀）
Thread.sleep(15000L);
System.out.println("當(dāng)前線(xiàn)程池線(xiàn)程數(shù)量為：" threadPoolExecutor.getPoolSize());
System.out.println("當(dāng)前線(xiàn)程池等待的數(shù)量為：" threadPoolExecutor.getQueue().size());
}
/**
* 1、線(xiàn)程池信息： 核心線(xiàn)程數(shù)量5，最大數(shù)量10，無(wú)界隊(duì)列，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest1() throws Exception {
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
testCommon(threadPoolExecutor);
// 預(yù)計(jì)結(jié)果：線(xiàn)程池線(xiàn)程數(shù)量為：5,超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
}
/**
* 2、 線(xiàn)程池信息： 核心線(xiàn)程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest2() throws Exception {
// 創(chuàng)建一個(gè) 核心線(xiàn)程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3 的線(xiàn)程池，也就是最大容納13個(gè)任務(wù)。
// 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,
10,
5,
TimeUnit.SECONDS,
new LinkedBlockingDeque<Runnable>(3),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.err.println("有任務(wù)被拒絕執(zhí)行了");
}
});
testCommon(threadPoolExecutor);
// 預(yù)計(jì)結(jié)果：
// 1、 5個(gè)任務(wù)直接分配線(xiàn)程開(kāi)始執(zhí)行
// 2、 3個(gè)任務(wù)進(jìn)入等待隊(duì)列
// 3、 隊(duì)列不夠用，臨時(shí)加開(kāi)5個(gè)線(xiàn)程來(lái)執(zhí)行任務(wù)(5秒沒(méi)活干就銷(xiāo)毀)
// 4、 隊(duì)列和線(xiàn)程池都滿(mǎn)了，剩下2個(gè)任務(wù)，沒(méi)資源了，被拒絕執(zhí)行。
// 5、 任務(wù)執(zhí)行，5秒后，如果無(wú)任務(wù)可執(zhí)行，銷(xiāo)毀臨時(shí)創(chuàng)建的5個(gè)線(xiàn)程
}
/**
* 3、 線(xiàn)程池信息： 核心線(xiàn)程數(shù)量5，最大數(shù)量5，無(wú)界隊(duì)列，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：5秒
*
* @throws Exception
*/
private void threadPoolExecutorTest3() throws Exception {
// 和Executors.newFixedThreadPool(int nThreads)一樣的
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
testCommon(threadPoolExecutor);
// 預(yù)計(jì)結(jié)：線(xiàn)程池線(xiàn)程數(shù)量為：5，超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
}
/**
* 4、 線(xiàn)程池信息： 核心線(xiàn)程數(shù)量0，最大數(shù)量Integer.MAX_VALUE，SynchronousQueue隊(duì)列，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：60秒
*
* @throws Exception
*/
private void threadPoolExecutorTest4() throws Exception {
/**
* SynchronousQueue，實(shí)際上它不是一個(gè)真正的隊(duì)列，因?yàn)樗粫?huì)為隊(duì)列中元素維護(hù)存儲(chǔ)空間。 與其他隊(duì)列不同的是，它維護(hù)一組線(xiàn)程，這些線(xiàn)程在等待著把元素加入或移出隊(duì)列。
* 在使用SynchronousQueue作為工作隊(duì)列的前提下，客戶(hù)端代碼向線(xiàn)程池提交任務(wù)時(shí)， 而線(xiàn)程池中又沒(méi)有空閑的線(xiàn)程能夠從SynchronousQueue隊(duì)列實(shí)例中取一個(gè)任務(wù)，
* 那么相應(yīng)的offer方法調(diào)用就會(huì)失?。慈蝿?wù)沒(méi)有被存入工作隊(duì)列）。 此時(shí)，ThreadPoolExecutor會(huì)新建一個(gè)新的工作者線(xiàn)程用于對(duì)這個(gè)入隊(duì)列失敗的任務(wù)進(jìn)行處理
* （假設(shè)此時(shí)線(xiàn)程池的大小還未達(dá)到其最大線(xiàn)程池大小maximumPoolSize）。
*/
// 和Executors.newCachedThreadPool()一樣的
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
testCommon(threadPoolExecutor);
// 預(yù)計(jì)結(jié)果：
// 1、 線(xiàn)程池線(xiàn)程數(shù)量為：15，超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
// 2、 所有任務(wù)執(zhí)行結(jié)束，60秒后，如果無(wú)任務(wù)可執(zhí)行，所有線(xiàn)程全部被銷(xiāo)毀，池的大小恢復(fù)為0
Thread.sleep(60000L);
System.out.println("60秒后，再看線(xiàn)程池中的數(shù)量：" threadPoolExecutor.getPoolSize());
}
/**
* 5、 定時(shí)執(zhí)行線(xiàn)程池信息：3秒后執(zhí)行，一次性任務(wù)，到點(diǎn)就執(zhí)行
* 核心線(xiàn)程數(shù)量5，最大數(shù)量Integer.MAX_VALUE，DelayedWorkQueue延時(shí)隊(duì)列，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：0秒
*
* @throws Exception
*/
private void threadPoolExecutorTest5() throws Exception {
// 和Executors.newScheduledThreadPool()一樣的
ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
threadPoolExecutor.schedule(
new Runnable() {
@Override
public void run() {
System.out.println("任務(wù)被執(zhí)行，現(xiàn)在時(shí)間：" System.currentTimeMillis());
}
},
3000,
TimeUnit.MILLISECONDS);
System.out.println(
"定時(shí)任務(wù)，提交成功，時(shí)間是："
System.currentTimeMillis()
", 當(dāng)前線(xiàn)程池中線(xiàn)程數(shù)量："
threadPoolExecutor.getPoolSize());
// 預(yù)計(jì)結(jié)果：任務(wù)在3秒后被執(zhí)行一次
}
/**
* 6、 定時(shí)執(zhí)行線(xiàn)程池信息：線(xiàn)程固定數(shù)量5 ，<br>
* 核心線(xiàn)程數(shù)量5，最大數(shù)量Integer.MAX_VALUE，DelayedWorkQueue延時(shí)隊(duì)列，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：0秒
*
* @throws Exception
*/
private void threadPoolExecutorTest6() throws Exception {
ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
/**
* 周期性執(zhí)行某一個(gè)任務(wù)，線(xiàn)程池提供了兩種調(diào)度方式，這里單獨(dú)演示一下。測(cè)試場(chǎng)景一樣。 測(cè)試場(chǎng)景：提交的任務(wù)需要3秒才能執(zhí)行完畢?？磧煞N不同調(diào)度方式的區(qū)別 效果1：
* 提交后，2秒后開(kāi)始第一次執(zhí)行，之后每間隔1秒，固定執(zhí)行一次(如果發(fā)現(xiàn)上次執(zhí)行還未完畢，則等待完畢，完畢后立刻執(zhí)行)。
* 也就是說(shuō)這個(gè)代碼中是，3秒鐘執(zhí)行一次（計(jì)算方式：每次執(zhí)行三秒，間隔時(shí)間1秒，執(zhí)行結(jié)束后馬上開(kāi)始下一次執(zhí)行，無(wú)需等待）
*/
threadPoolExecutor.scheduleAtFixedRate(
new Runnable() {
@Override
public void run() {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任務(wù)-1 被執(zhí)行，現(xiàn)在時(shí)間：" System.currentTimeMillis());
}
},
2000,
1000,
TimeUnit.MILLISECONDS);
/**
* 效果2：提交后，2秒后開(kāi)始第一次執(zhí)行，之后每間隔1秒，固定執(zhí)行一次(如果發(fā)現(xiàn)上次執(zhí)行還未完畢，則等待完畢，等上一次執(zhí)行完畢后再開(kāi)始計(jì)時(shí)，等待1秒)。
* 也就是說(shuō)這個(gè)代碼鐘的效果看到的是：4秒執(zhí)行一次。 （計(jì)算方式：每次執(zhí)行3秒，間隔時(shí)間1秒，執(zhí)行完以后再等待1秒，所以是 3 1）
*/
threadPoolExecutor.scheduleWithFixedDelay(
new Runnable() {
@Override
public void run() {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任務(wù)-2 被執(zhí)行，現(xiàn)在時(shí)間：" System.currentTimeMillis());
}
},
2000,
1000,
TimeUnit.MILLISECONDS);
}
private ThreadPoolExecutor testShutdownCommon() {
// 創(chuàng)建一個(gè) 核心線(xiàn)程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3的線(xiàn)程池，也就是最大容納13個(gè)任務(wù)。
// 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,
10,
5,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(3),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.err.println("有任務(wù)被拒絕執(zhí)行了");
}
});
// 測(cè)試： 提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù)，看超過(guò)大小的2個(gè)，對(duì)應(yīng)的處理情況
for (int i = 0; i < 15; i ) {
int n = i;
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
try {
System.out.println("開(kāi)始執(zhí)行：" n);
Thread.sleep(3000L);
System.err.println("執(zhí)行結(jié)束:" n);
} catch (InterruptedException e) {
System.out.println("異常：" e.getMessage());
}
}
});
System.out.println("任務(wù)提交成功 :" i);
}
return threadPoolExecutor;
}
/**
* 7、 終止線(xiàn)程：線(xiàn)程池信息： 核心線(xiàn)程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest7() throws Exception {
ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
// 1秒后終止線(xiàn)程池
Thread.sleep(1000L);
threadPoolExecutor.shutdown();
// 再次提交提示失敗
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
System.out.println("追加一個(gè)任務(wù)");
}
});
// 結(jié)果分析
// 1、 10個(gè)任務(wù)被執(zhí)行，3個(gè)任務(wù)進(jìn)入隊(duì)列等待，2個(gè)任務(wù)被拒絕執(zhí)行
// 2、調(diào)用shutdown后，不接收新的任務(wù)，等待13任務(wù)執(zhí)行結(jié)束
// 3、 追加的任務(wù)在線(xiàn)程池關(guān)閉后，無(wú)法再提交，會(huì)被拒絕執(zhí)行
}
/**
* 8、 立刻終止線(xiàn)程：線(xiàn)程池信息： 核心線(xiàn)程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線(xiàn)程數(shù)量的線(xiàn)程存活時(shí)間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest8() throws Exception {
// 創(chuàng)建一個(gè) 核心線(xiàn)程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3 的線(xiàn)程池，也就是最大容納13個(gè)任務(wù)。
// 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
// 1秒后終止線(xiàn)程池
Thread.sleep(1000L);
List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
// 再次提交提示失敗
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
System.out.println("追加一個(gè)任務(wù)");
}
});
System.out.println("未結(jié)束的任務(wù)有：" shutdownNow.size());
// 結(jié)果分析
// 1、 10個(gè)任務(wù)被執(zhí)行，3個(gè)任務(wù)進(jìn)入隊(duì)列等待，2個(gè)任務(wù)被拒絕執(zhí)行
// 2、調(diào)用shutdownnow后，隊(duì)列中的3個(gè)線(xiàn)程不再執(zhí)行，10個(gè)線(xiàn)程被終止
// 3、 追加的任務(wù)在線(xiàn)程池關(guān)閉后，無(wú)法再提交，會(huì)被拒絕執(zhí)行
}
public static void main(String[] args) throws Exception {
Demo7 demo7 = new Demo7();
// demo7.threadPoolExecutorTest1();
// demo7.threadPoolExecutorTest2();
// demo7.threadPoolExecutorTest3();
// demo7.threadPoolExecutorTest4();
// demo7.threadPoolExecutorTest5();
// demo7.threadPoolExecutorTest6();
// demo7.threadPoolExecutorTest7();
demo7.threadPoolExecutorTest8();
}
}
來(lái)源：https://www./content-1-321551.html
|
