Java如何自定義線程池中隊列
背景
業務交互的過程中涉及到瞭很多關於SFTP下載的問題,因此在代碼中定義瞭一些線程池,使用中發現瞭一些問題,
代碼類似如下所示:
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(); } } } } }
如上面的代碼所示,定義瞭一個初始容量為2,最大容量為100,隊列長度為10的線程池,期待的運行結果為:
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
期待十個線程都可以運行,但實際的執行效果如下:
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
對比可以看出,用上面的方式定義線程池,最終隻有兩個線程可以運行,即線程池的初始容量大小。其餘線程都被阻塞到瞭隊列ArrayBlockingQueue<>(10)
問題分析
我們知道,Executors框架提供瞭幾種常見的線程池分別為:
- newCachedThreadPool創建一個可緩存線程池,如果線程池長度超過處理需要,可靈活回收空閑線程,若無可回收,則新建線程。
- newFixedThreadPool 創建一個定長線程池,可控制線程最大並發數,超出的線程會在隊列中等待。
- newScheduledThreadPool 創建一個定長線程池,支持定時及周期性任務執行。
- newSingleThreadExecutor 創建一個單線程化的線程池,它隻會用唯一的工作線程來執行任務,保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。
如果將代碼中自定義的線程池改為 :
private static ExecutorService es = Executors.newCachedThreadPool();
運行發現,提交的十個線程都可以運行
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>()); }
通過對比發現,newCachedThreadPool使用的是 SynchronousQueue<>()而我們使用的是ArrayBlockingQueue<>(10) 因此可以很容易的發現問題出在隊列上。
問題解決
將ArrayBlockingQueue改為SynchronousQueue 問題解決,代碼如下:
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(); } } } } }
總結
兩個隊列的UML關系圖
從圖上我們可以看到,兩個隊列都繼承瞭AbstractQueue實現瞭BlockingQueue接口,因此功能應該相似
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類似於一個傳遞通道,隻是通過他傳遞某個元素,並沒有任何容量,隻有當第一個元素被取走,才能在給隊列添加元素。
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從定義來看就是一個普通的隊列,先入先出,當隊列為空時,獲取數據的線程會被阻塞,當隊列滿時,添加隊列的線程會被阻塞,直到隊列可用。
分析
從上面隊列的定義中可以看出,導致線程池沒有按照預期運行的原因不是因為隊列的問題,應該是關於線程池在提交任務時,從隊列取數據的方式不同導致的。
jdk源碼中關於線程池隊列的說明
* <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>
從說明中可以看到,如果正在運行的線程數必初始容量corePoolSize小,那麼Executor會從創建一個新線程去執行任務,如果正在執行的線程數必corePoolSize大,那麼Executor會將新提交的任務放到阻塞隊列,除非當隊列的個數超過瞭隊列的最大長度maxmiumPooSize。
從源碼中找到關於提交任務的方法:
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; }
從源碼中看到 subimit實際上是調用瞭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); }
源碼中可以看出,提交任務時,首先會判斷正在執行的線程數是否小於corePoolSize,如果條件成立那麼會直接創建線程並執行任務。如果條件不成立,且隊列沒有滿,那麼將任務放到隊列,如果條件不成立但是隊列滿瞭,那麼同樣也新創建線程並執行任務。
到此這篇關於Java如何自定義線程池中隊列的文章就介紹到這瞭,更多相關Java 隊列內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Java如何固定大小的線程池
- Java並發編程之Executor接口的使用
- 詳解Java ThreadPoolExecutor的拒絕策略
- Java線程池的簡單使用方法實例教程
- java 線程池keepAliveTime的含義說明