Java中線(xiàn)程池自定義實(shí)現(xiàn)詳解
前言
最初使用線(xiàn)程池的時(shí)候,網(wǎng)上的文章告訴我說(shuō)線(xiàn)程池可以線(xiàn)程復(fù)用,提高線(xiàn)程的創(chuàng)建效率。從此我的腦海中便為線(xiàn)程池打上了一個(gè)標(biāo)簽——線(xiàn)程池可以做到線(xiàn)程的復(fù)用。但是我總以為線(xiàn)程的復(fù)用是指在創(chuàng)建出來(lái)的線(xiàn)程可以多次的更換run()方法的內(nèi)容,來(lái)達(dá)到線(xiàn)程復(fù)用的目的,于是我嘗試了一下.同一個(gè)線(xiàn)程調(diào)用多次,然后使run的內(nèi)容不一樣,但是我發(fā)現(xiàn)我錯(cuò)了,一個(gè)線(xiàn)程第一次運(yùn)行是沒(méi)問(wèn)題的,當(dāng)再次調(diào)用start方法是會(huì)拋出異常(java.lang.IllegalThreadStateException)。
線(xiàn)程為什么不能多次調(diào)用start方法
從源碼可以得知,調(diào)用start方法時(shí),程序還會(huì)判斷當(dāng)前的線(xiàn)程狀態(tài)
這里又引申出另一個(gè)問(wèn)題,線(xiàn)程到底有幾種狀態(tài)
年輕的時(shí)候背八股文時(shí),只是說(shuō)五種狀態(tài),這五種狀態(tài)也不知道是哪里來(lái)的,不知道有沒(méi)有人和我一樣,當(dāng)初只是知其然不知其所以然。貼出源碼來(lái):
public enum State { /** * Thread state for a thread which has not yet started. */ NEW, // 新建 /** * Thread state for a runnable thread. A thread in the runnable * state is executing in the Java virtual machine but it may * be waiting for other resources from the operating system * such as processor. */ RUNNABLE, // 運(yùn)行中 /** * Thread state for a thread blocked waiting for a monitor lock. * A thread in the blocked state is waiting for a monitor lock * to enter a synchronized block/method or * reenter a synchronized block/method after calling * {@link Object#wait() Object.wait}. */ BLOCKED, // 阻塞 /** * Thread state for a waiting thread. * A thread is in the waiting state due to calling one of the * following methods: * <ul> * <li>{@link Object#wait() Object.wait} with no timeout</li> * <li>{@link #join() Thread.join} with no timeout</li> * <li>{@link LockSupport#park() LockSupport.park}</li> * </ul> * * <p>A thread in the waiting state is waiting for another thread to * perform a particular action. * * For example, a thread that has called <tt>Object.wait()</tt> * on an object is waiting for another thread to call * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on * that object. A thread that has called <tt>Thread.join()</tt> * is waiting for a specified thread to terminate. */ WAITING, // 等待 /** * Thread state for a waiting thread with a specified waiting time. * A thread is in the timed waiting state due to calling one of * the following methods with a specified positive waiting time: * <ul> * <li>{@link #sleep Thread.sleep}</li> * <li>{@link Object#wait(long) Object.wait} with timeout</li> * <li>{@link #join(long) Thread.join} with timeout</li> * <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li> * <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li> * </ul> */ TIMED_WAITING, // 定時(shí)等待 /** * Thread state for a terminated thread. * The thread has completed execution. */ TERMINATED; // 結(jié)束狀態(tài) }
綜上,其實(shí)線(xiàn)程的狀態(tài)有六種:
- NEW 新建狀態(tài),一般通過(guò)Thread thread = new Thread(runable);此時(shí)的線(xiàn)程屬于新建狀態(tài)。
- RUNABLE 可運(yùn)行狀態(tài),當(dāng)調(diào)用start時(shí),線(xiàn)程進(jìn)入RUNNABLE狀態(tài),該狀態(tài)其實(shí)還包含兩個(gè)狀態(tài),一種是被cpu選中正在運(yùn)行中,另一種是未被cpu選中,處于就緒狀態(tài)。
- BLOCKED 阻塞狀態(tài), 一般可以通過(guò)調(diào)用sleep()方法來(lái)進(jìn)入阻塞狀態(tài),此時(shí)線(xiàn)程沒(méi)有釋放鎖資源,sleep到期時(shí),繼續(xù)進(jìn)入Runable狀態(tài)
- WAITING 等待狀態(tài), 一般可以通過(guò)調(diào)用wait()方法來(lái)進(jìn)入等待狀態(tài),此時(shí)釋放cpu,cpu去干其他事情,需要調(diào)用noitfy方法喚醒,喚醒后的線(xiàn)程為RUNABLE狀態(tài)。
- TIMED_WAIRING 定時(shí)等待, 一般可以通過(guò)wait(long)方法進(jìn)入定時(shí)等待?;旧贤琖AITING.
- TERMINATED 結(jié)束狀態(tài),RUNCABLE運(yùn)行正常結(jié)束的線(xiàn)程的狀態(tài)就是TERMINATED
可以看出八股文不能亂背,之前傻呵呵背的八股文很有可能是錯(cuò)誤的,比如線(xiàn)程的運(yùn)行中狀態(tài)(RUNNING),其實(shí)這個(gè)狀態(tài)根本不存在,RUNABLE狀態(tài)就已經(jīng)包含了RUNNNING狀態(tài)了。
再回到標(biāo)題的問(wèn)題,為什么不能多次調(diào)用start方法,原因其實(shí)源碼的注釋上已經(jīng)說(shuō)明了,
/** * This method is not invoked for the main method thread or "system" * group threads created/set up by the VM. Any new functionality added * to this method in the future may have to also be added to the VM. * * A zero status value corresponds to state "NEW". */
0狀態(tài)對(duì)應(yīng)的是NEW,也就是說(shuō)只有新建狀態(tài)的線(xiàn)程才能調(diào)用start方法,其他狀態(tài)的線(xiàn)程調(diào)用就會(huì)拋出異常,而一般第二次調(diào)用時(shí),線(xiàn)程狀態(tài)肯定不是new狀態(tài)了。因此不可以多次調(diào)用。
線(xiàn)程池到底是如何復(fù)用的
經(jīng)過(guò)多次的反復(fù)調(diào)試,原理其實(shí)很簡(jiǎn)單,比如以下代碼:
public void testThreadPool() { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 3, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue(3)); threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); for (int i=0; i<5; i++) { threadPoolExecutor.submit(new Runnable() { @Override public void run() { ThreadUtils.doSleep(10000L); System.out.println(Thread.currentThread().getName() + "--運(yùn)行"); } }); } threadPoolExecutor.shutdown(); }
其中循環(huán)往threadPoolExecutor中添加的是自定義的業(yè)務(wù)任務(wù)。而真正去運(yùn)行任務(wù)的是線(xiàn)程池中新建的一個(gè)線(xiàn)程。因此這里的復(fù)用指的是線(xiàn)程池創(chuàng)建出來(lái)得這個(gè)線(xiàn)程,這個(gè)線(xiàn)程并不會(huì)銷(xiāo)毀,而是循環(huán)去隊(duì)列中獲取任務(wù)。千萬(wàn)不可理解為線(xiàn)程池復(fù)用的線(xiàn)程是使用者自定義的那個(gè)業(yè)務(wù)任務(wù)。具體的復(fù)用最核心的代碼就是下面這段:
while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } }
這段代碼是runworker中的一段代碼,線(xiàn)程就是通過(guò)循環(huán)去獲取隊(duì)列中的任務(wù)來(lái)達(dá)到線(xiàn)程復(fù)用的,前臺(tái)創(chuàng)建多個(gè)runable對(duì)象,將任務(wù)放到runable中,然后將runable放到隊(duì)列中,線(xiàn)程池創(chuàng)建線(xiàn)程,線(xiàn)程持續(xù)循環(huán)獲取隊(duì)列中的任務(wù)。這就是線(xiàn)程池的實(shí)現(xiàn)邏輯。
下面嘗試自己去實(shí)現(xiàn)一個(gè)線(xiàn)程池:該線(xiàn)程只是為了模擬線(xiàn)程池的運(yùn)行,并未做線(xiàn)程安全的考慮,也未做非核心線(xiàn)程超時(shí)回收等功能。
package com.cz.lock.distributed.impl.redis; import java.util.List; import java.util.concurrent.*; /** * @program: Reids * @description: 自定義線(xiàn)程池 * @author: Cheng Zhi * @create: 2023-02-28 09:28 **/ public class JefThreadPoolExecutor extends AbstractExecutorService { /** * 使用隊(duì)列來(lái)保存現(xiàn)有的worker */ private final BlockingQueue<Worker> workers = new LinkedBlockingQueue<Worker>(); private static int coreThreadCount = 5; private static int maxThreadCount = 10; private static int defaultQueueSize = maxThreadCount * 5; private static BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(defaultQueueSize); /** * 默認(rèn)線(xiàn)程池 */ JefThreadPoolExecutor() { this(coreThreadCount, maxThreadCount, blockingQueue); } /** * 可以自定義的線(xiàn)程池 * @param coreThreadCount * @param maxThreadCount * @param blockingQueue */ JefThreadPoolExecutor(int coreThreadCount, int maxThreadCount, BlockingQueue blockingQueue) { this.blockingQueue = blockingQueue; this.coreThreadCount = coreThreadCount; this.maxThreadCount = maxThreadCount; } @Override public void shutdown() { } @Override public List<Runnable> shutdownNow() { return null; } @Override public boolean isShutdown() { return false; } @Override public boolean isTerminated() { return false; } @Override public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { return false; } @Override public void execute(Runnable command) { int currentWorkCount = workers.size(); // 當(dāng)前創(chuàng)建的線(xiàn)程總數(shù) if (currentWorkCount < coreThreadCount) { // 如果當(dāng)前線(xiàn)程總數(shù)小于核心線(xiàn)程數(shù),則新建線(xiàn)程 Worker worker = new Worker(command); final Thread thread = worker.thread; thread.start(); addWorker(worker); return; } if (!blockingQueue.offer(command) && currentWorkCount <= maxThreadCount) { // 隊(duì)列可以正常放入則返回true,如果滿(mǎn)了返回false // 隊(duì)列如果滿(mǎn)了,需要?jiǎng)?chuàng)建新的線(xiàn)程 Worker worker = new Worker(command); final Thread thread = worker.thread; thread.start(); addWorker(worker); return; } else if (currentWorkCount > maxThreadCount){ System.out.println("線(xiàn)程池滿(mǎn)了....沒(méi)有多余的線(xiàn)程了"); } } public void addWorker(Worker worker) { workers.add(worker); } public Runnable getTask() { Runnable poll = blockingQueue.poll(); return poll; } public void runWorker(Worker worker) { Runnable task = worker.firstTask; // 獲取到new Worker時(shí)傳入的那個(gè)任務(wù),并在下面運(yùn)行 if (task != null) { task.run(); } worker.firstTask = null; // 循環(huán)從隊(duì)列中獲取任務(wù)處理 while((task = getTask()) != null) { task.run(); } } /** * 匿名內(nèi)部類(lèi) */ private class Worker implements Runnable{ volatile int state = 0; public Runnable firstTask; final Thread thread; public Worker(Runnable firstTask) { this.firstTask = firstTask; thread = new Thread(this); } @Override public void run() { runWorker(this); } } }
使用方式:
/** * 使用默認(rèn)配置 */ public static void singleThreadPoolExecutor() { JefThreadPoolExecutor jefThreadPoolExecutor = new JefThreadPoolExecutor(); for (int i=0; i<10; i++) { jefThreadPoolExecutor.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + "--運(yùn)行"); } }); } } /** * 自定義配置 */ public static void diyThreadPoolExecutor() { JefThreadPoolExecutor jefThreadPoolExecutor = new JefThreadPoolExecutor(2, 10, new ArrayBlockingQueue(50)); for (int i=0; i<500; i++) { jefThreadPoolExecutor.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + "--運(yùn)行"); } }); } }
以上就是Java中線(xiàn)程池自定義實(shí)現(xiàn)詳解的詳細(xì)內(nèi)容,更多關(guān)于Java線(xiàn)程池的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
解決@Test注解在Maven工程的Test.class類(lèi)中無(wú)法使用的問(wèn)題
這篇文章主要介紹了解決@Test注解在Maven工程的Test.class類(lèi)中無(wú)法使用的問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-03-03使用springboot不自動(dòng)初始化數(shù)據(jù)庫(kù)連接池
這篇文章主要介紹了使用springboot不自動(dòng)初始化數(shù)據(jù)庫(kù)連接池,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-09-09SpringBoot?替換?if?的參數(shù)校驗(yàn)示例代碼
Spring?Validation是對(duì)hibernate?validation的二次封裝,用于支持spring?mvc參數(shù)自動(dòng)校驗(yàn),接下來(lái),我們以spring-boot項(xiàng)目為例,介紹Spring?Validation的使用,需要的朋友可以參考下2022-12-12Java使用Canal同步MySQL數(shù)據(jù)到Redis
在現(xiàn)代微服務(wù)架構(gòu)中,數(shù)據(jù)同步是一個(gè)常見(jiàn)的需求,特別是將?MySQL?數(shù)據(jù)實(shí)時(shí)同步到?Redis,下面我們就來(lái)看看Java如何使用Canal同步MySQL數(shù)據(jù)到Redis吧2024-11-11springboot打包jar中沒(méi)有主清單屬性問(wèn)題
這篇文章主要介紹了springboot打包jar中沒(méi)有主清單屬性問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-12-12idea中增強(qiáng)for循環(huán)提示unexpected token問(wèn)題
這篇文章主要介紹了idea中增強(qiáng)for循環(huán)提示unexpected token問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-01-01Java中String.split()的最詳細(xì)源碼解讀及注意事項(xiàng)
以前經(jīng)常使用String.split()方法,但是從來(lái)沒(méi)有注意,下面這篇文章主要給大家介紹了關(guān)于Java中String.split()最詳細(xì)源碼解讀及注意事項(xiàng)的相關(guān)資料,文中通過(guò)圖文介紹的非常詳細(xì),需要的朋友可以參考下2022-07-07Java數(shù)據(jù)結(jié)構(gòu)篇之實(shí)現(xiàn)二叉搜索樹(shù)的核心方法
二叉搜索樹(shù)是一種常用的數(shù)據(jù)結(jié)構(gòu),它是一棵二叉樹(shù),且每個(gè)節(jié)點(diǎn)的值都大于其左子樹(shù)中任何節(jié)點(diǎn)的值,而小于其右子樹(shù)中任何節(jié)點(diǎn)的值,這篇文章主要給大家介紹了關(guān)于Java數(shù)據(jù)結(jié)構(gòu)篇之實(shí)現(xiàn)二叉搜索樹(shù)的核心方法,需要的朋友可以參考下2023-12-12