首頁 > 軟體

Java如何自定義執行緒池中佇列

2022-07-07 14:01:17

背景

業務互動的過程中涉及到了很多關於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 佇列內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com!


IT145.com E-mail:sddin#qq.com