Java怎么自定義線程池中隊列

本篇內(nèi)容介紹了“Java怎么自定義線程池中隊列”的有關(guān)知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細閱讀，能夠?qū)W有所成！

背景

業(yè)務(wù)交互的過程中涉及到了很多關(guān)于SFTP下載的問題，因此在代碼中定義了一些線程池，使用中發(fā)現(xià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();
                }
            }
        }
    }
}

如上面的代碼所示，定義了一個初始容量為2，最大容量為100，隊列長度為10的線程池，期待的運行結(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

期待十個線程都可以運行，但實際的執(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

對比可以看出，用上面的方式定義線程池，最終只有兩個線程可以運行，即線程池的初始容量大小。其余線程都被阻塞到了隊列ArrayBlockingQueue<>(10)

問題分析

我們知道，Executors框架提供了幾種常見的線程池分別為：

• newCachedThreadPool創(chuàng)建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。

• newFixedThreadPool 創(chuàng)建一個定長線程池，可控制線程最大并發(fā)數(shù)，超出的線程會在隊列中等待。

• newScheduledThreadPool 創(chuàng)建一個定長線程池，支持定時及周期性任務(wù)執(zhí)行。

• newSingleThreadExecutor 創(chuàng)建一個單線程化的線程池，它只會用唯一的工作線程來執(zhí)行任務(wù)，保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級)執(zhí)行。

如果將代碼中自定義的線程池改為 ：

private static ExecutorService es = Executors.newCachedThreadPool();

運行發(fā)現(xià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>());
}

通過對比發(fā)現(xiàn)，newCachedThreadPool使用的是 SynchronousQueue<>()而我們使用的是ArrayBlockingQueue<>(10) 因此可以很容易的發(fā)現(xiàn)問題出在隊列上。

問題解決

將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();
                }
            }
        }
    }
}

總結(jié)

兩個隊列的UML關(guān)系圖

從圖上我們可以看到，兩個隊列都繼承了AbstractQueue實現(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類似于一個傳遞通道,只是通過他傳遞某個元素，并沒有任何容量，只有當(dāng)?shù)谝粋€元素被取走，才能在給隊列添加元素。

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從定義來看就是一個普通的隊列，先入先出，當(dāng)隊列為空時，獲取數(shù)據(jù)的線程會被阻塞，當(dāng)隊列滿時，添加隊列的線程會被阻塞，直到隊列可用。

分析

從上面隊列的定義中可以看出，導(dǎo)致線程池沒有按照預(yù)期運行的原因不是因為隊列的問題，應(yīng)該是關(guān)于線程池在提交任務(wù)時，從隊列取數(shù)據(jù)的方式不同導(dǎo)致的。

jdk源碼中關(guān)于線程池隊列的說明

* <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>

從說明中可以看到，如果正在運行的線程數(shù)必初始容量corePoolSize小，那么Executor會從創(chuàng)建一個新線程去執(zhí)行任務(wù)，如果正在執(zhí)行的線程數(shù)必corePoolSize大，那么Executor會將新提交的任務(wù)放到阻塞隊列，除非當(dāng)隊列的個數(shù)超過了隊列的最大長度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實際上是調(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ù)時，首先會判斷正在執(zhí)行的線程數(shù)是否小于corePoolSize，如果條件成立那么會直接創(chuàng)建線程并執(zhí)行任務(wù)。如果條件不成立，且隊列沒有滿，那么將任務(wù)放到隊列，如果條件不成立但是隊列滿了，那么同樣也新創(chuàng)建線程并執(zhí)行任務(wù)。

“Java怎么自定義線程池中隊列”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實用文章！