Java如何自定義線程池中隊(duì)列
背景
業(yè)務(wù)交互的過(guò)程中涉及到了很多關(guān)于SFTP下載的問(wèn)題,因此在代碼中定義了一些線程池,使用中發(fā)現(xiàn)了一些問(wèn)題,
代碼類似如下所示:
public class ExecutorTest { private static ExecutorService es = new ThreadPoolExecutor(2, 100, 1000, TimeUnit.MILLISECONDS , new ArrayBlockingQueue<>(10)); public static void main(String[] args) { for (int i = 0; i < 10; i++) { es.submit(new MyThread()); } } static class MyThread implements Runnable { @Override public void run() { for (; ; ) { System.out.println("Thread name=" + Thread.currentThread().getName()); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } } } } }
如上面的代碼所示,定義了一個(gè)初始容量為2,最大容量為100,隊(duì)列長(zhǎng)度為10的線程池,期待的運(yùn)行結(jié)果為:
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-3
Thread name=pool-1-thread-4
Thread name=pool-1-thread-5
Thread name=pool-1-thread-6
Thread name=pool-1-thread-7
Thread name=pool-1-thread-8
Thread name=pool-1-thread-9
Thread name=pool-1-thread-10
Thread name=pool-1-thread-3
Thread name=pool-1-thread-5
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
Thread name=pool-1-thread-4
Thread name=pool-1-thread-10
Thread name=pool-1-thread-7
Thread name=pool-1-thread-6
Thread name=pool-1-thread-9
Thread name=pool-1-thread-8
Thread name=pool-1-thread-3
Thread name=pool-1-thread-4
Thread name=pool-1-thread-1
Thread name=pool-1-thread-5
Thread name=pool-1-thread-2
Thread name=pool-1-thread-8
Thread name=pool-1-thread-6
Thread name=pool-1-thread-7
Thread name=pool-1-thread-9
Thread name=pool-1-thread-10
期待十個(gè)線程都可以運(yùn)行,但實(shí)際的執(zhí)行效果如下:
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
Thread name=pool-1-thread-2
Thread name=pool-1-thread-1
對(duì)比可以看出,用上面的方式定義線程池,最終只有兩個(gè)線程可以運(yùn)行,即線程池的初始容量大小。其余線程都被阻塞到了隊(duì)列ArrayBlockingQueue<>(10)
問(wèn)題分析
我們知道,Executors框架提供了幾種常見(jiàn)的線程池分別為:
- newCachedThreadPool創(chuàng)建一個(gè)可緩存線程池,如果線程池長(zhǎng)度超過(guò)處理需要,可靈活回收空閑線程,若無(wú)可回收,則新建線程。
- newFixedThreadPool 創(chuàng)建一個(gè)定長(zhǎng)線程池,可控制線程最大并發(fā)數(shù),超出的線程會(huì)在隊(duì)列中等待。
- newScheduledThreadPool 創(chuàng)建一個(gè)定長(zhǎng)線程池,支持定時(shí)及周期性任務(wù)執(zhí)行。
- newSingleThreadExecutor 創(chuàng)建一個(gè)單線程化的線程池,它只會(huì)用唯一的工作線程來(lái)執(zhí)行任務(wù),保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級(jí))執(zhí)行。
如果將代碼中自定義的線程池改為 :
private static ExecutorService es = Executors.newCachedThreadPool();
運(yùn)行發(fā)現(xiàn),提交的十個(gè)線程都可以運(yùn)行
Executors.newCachedThreadPool()的源碼如下:
/** * Creates a thread pool that creates new threads as needed, but * will reuse previously constructed threads when they are * available. These pools will typically improve the performance * of programs that execute many short-lived asynchronous tasks. * Calls to {@code execute} will reuse previously constructed * threads if available. If no existing thread is available, a new * thread will be created and added to the pool. Threads that have * not been used for sixty seconds are terminated and removed from * the cache. Thus, a pool that remains idle for long enough will * not consume any resources. Note that pools with similar * properties but different details (for example, timeout parameters) * may be created using {@link ThreadPoolExecutor} constructors. * * @return the newly created thread pool */ public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
通過(guò)對(duì)比發(fā)現(xiàn),newCachedThreadPool使用的是 SynchronousQueue<>()而我們使用的是ArrayBlockingQueue<>(10) 因此可以很容易的發(fā)現(xiàn)問(wèn)題出在隊(duì)列上。
問(wèn)題解決
將ArrayBlockingQueue改為SynchronousQueue 問(wèn)題解決,代碼如下:
public class ExecutorTest { private static ExecutorService es = new ThreadPoolExecutor(2, 100, 1000, TimeUnit.MILLISECONDS , new SynchronousQueue<>()); private static ExecutorService es2 = Executors.newCachedThreadPool(); public static void main(String[] args) { for (int i = 0; i < 10; i++) { es.submit(new MyThread()); } } static class MyThread implements Runnable { @Override public void run() { for (; ; ) { System.out.println("Thread name=" + Thread.currentThread().getName()); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } } } } }
總結(jié)
兩個(gè)隊(duì)列的UML關(guān)系圖
從圖上我們可以看到,兩個(gè)隊(duì)列都繼承了AbstractQueue實(shí)現(xiàn)了BlockingQueue接口,因此功能應(yīng)該相似
SynchronousQueue的定義
* <p>Synchronous queues are similar to rendezvous channels used in * CSP and Ada. They are well suited for handoff designs, in which an * object running in one thread must sync up with an object running * in another thread in order to hand it some information, event, or * task.
SynchronousQueue類似于一個(gè)傳遞通道,只是通過(guò)他傳遞某個(gè)元素,并沒(méi)有任何容量,只有當(dāng)?shù)谝粋€(gè)元素被取走,才能在給隊(duì)列添加元素。
ArrayBlockingQueue的定義
* A bounded {@linkplain BlockingQueue blocking queue} backed by an * array. This queue orders elements FIFO (first-in-first-out). The * <em>head</em> of the queue is that element that has been on the * queue the longest time. The <em>tail</em> of the queue is that * element that has been on the queue the shortest time. New elements * are inserted at the tail of the queue, and the queue retrieval * operations obtain elements at the head of the queue.
ArrayBlockingQueue從定義來(lái)看就是一個(gè)普通的隊(duì)列,先入先出,當(dāng)隊(duì)列為空時(shí),獲取數(shù)據(jù)的線程會(huì)被阻塞,當(dāng)隊(duì)列滿時(shí),添加隊(duì)列的線程會(huì)被阻塞,直到隊(duì)列可用。
分析
從上面隊(duì)列的定義中可以看出,導(dǎo)致線程池沒(méi)有按照預(yù)期運(yùn)行的原因不是因?yàn)殛?duì)列的問(wèn)題,應(yīng)該是關(guān)于線程池在提交任務(wù)時(shí),從隊(duì)列取數(shù)據(jù)的方式不同導(dǎo)致的。
jdk源碼中關(guān)于線程池隊(duì)列的說(shuō)明
* <dt>Queuing</dt> * * <dd>Any {@link BlockingQueue} may be used to transfer and hold * submitted tasks. The use of this queue interacts with pool sizing: * * <ul> * * <li> If fewer than corePoolSize threads are running, the Executor * always prefers adding a new thread * rather than queuing.</li> * * <li> If corePoolSize or more threads are running, the Executor * always prefers queuing a request rather than adding a new * thread.</li> * * <li> If a request cannot be queued, a new thread is created unless * this would exceed maximumPoolSize, in which case, the task will be * rejected.</li>
從說(shuō)明中可以看到,如果正在運(yùn)行的線程數(shù)必初始容量corePoolSize小,那么Executor會(huì)從創(chuàng)建一個(gè)新線程去執(zhí)行任務(wù),如果正在執(zhí)行的線程數(shù)必corePoolSize大,那么Executor會(huì)將新提交的任務(wù)放到阻塞隊(duì)列,除非當(dāng)隊(duì)列的個(gè)數(shù)超過(guò)了隊(duì)列的最大長(zhǎng)度maxmiumPooSize。
從源碼中找到關(guān)于提交任務(wù)的方法:
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; }
從源碼中看到 subimit實(shí)際上是調(diào)用了execute方法
execute方法的源碼:
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
源碼中可以看出,提交任務(wù)時(shí),首先會(huì)判斷正在執(zhí)行的線程數(shù)是否小于corePoolSize,如果條件成立那么會(huì)直接創(chuàng)建線程并執(zhí)行任務(wù)。如果條件不成立,且隊(duì)列沒(méi)有滿,那么將任務(wù)放到隊(duì)列,如果條件不成立但是隊(duì)列滿了,那么同樣也新創(chuàng)建線程并執(zhí)行任務(wù)。
到此這篇關(guān)于Java如何自定義線程池中隊(duì)列的文章就介紹到這了,更多相關(guān)Java 隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
IDEA中@Autowired自動(dòng)注入MyBatis?Mapper報(bào)紅警告的幾種解決方法
這篇文章主要介紹了IDEA中@Autowired自動(dòng)注入MyBatis?Mapper報(bào)紅警告的幾種解決方法2024-02-02詳解mybatis 批量更新數(shù)據(jù)兩種方法效率對(duì)比
這篇文章主要介紹了詳解mybatis 批量更新數(shù)據(jù)兩種方法效率對(duì)比,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-02-02SpringBoot實(shí)現(xiàn)根據(jù)手機(jī)號(hào)獲取歸屬地
這篇文章主要為大家詳細(xì)介紹了SpringBoot如何實(shí)現(xiàn)根據(jù)手機(jī)號(hào)獲取歸屬地,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-12-12SpringSecurity自定義資源攔截規(guī)則及登錄界面跳轉(zhuǎn)問(wèn)題
這篇文章主要介紹了SpringSecurity自定義資源攔截規(guī)則及登錄界面跳轉(zhuǎn)問(wèn)題,我們想要自定義認(rèn)證邏輯,就需要?jiǎng)?chuàng)建一些原來(lái)不存在的bean,這個(gè)時(shí)候就可以使@ConditionalOnMissingBean注解,本文給大家介紹的非常詳細(xì),需要的朋友參考下吧2023-12-12基于Mybatis Plus實(shí)現(xiàn)多表分頁(yè)查詢的示例代碼
這篇文章主要介紹了基于Mybatis Plus實(shí)現(xiàn)多表分頁(yè)查詢的示例代碼,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12