首頁 >Java >java教程 >Java 執行緒池的建立流程分析

Java 執行緒池的建立流程分析

黄舟
黄舟原創
2017-02-23 10:44:401392瀏覽

最近在改進專案的並發功能,但開發起來磕磕碰碰的。看好多資料,總算加深了認識。於是打算配合查看原始碼,總結並發程式設計的原理。

準備從用得最多的執行緒池開始,圍繞著創建、執行、關閉認識執行緒池整個生命週期的實作原理。後續再研究原子變數、並發容器、阻塞佇列、同步工具、鎖等主題。 java.util.concurrent裡的並發工具用起來不難,但不能只是會用,我們要read the fucking source code,哈哈。順便說聲,我用的JDK是1.8。

Executor框架

Executor是一套執行緒池管理框架,介面裡只有一個方法execute,執行Runnable任務。 ExecutorService介面擴展了Executor,增加了執行緒生命週期的管理,提供任務終止、傳回任務結果等方法。 AbstractExecutorService實作了ExecutorService,提供例如submit方法的預設實作邏輯。

然後到今天的主題ThreadPoolExecutor,繼承了AbstractExecutorService,提供線程池的具體實作。

建構方法

下面是ThreadPoolExecutor最普通的建構函數,最多有七個參數。具體程式碼不貼了,只是一些參數校驗設定的語句。

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
    }

corePoolSize是執行緒池的目標大小,也就是執行緒池剛剛建立起來,還沒有任務要執行時的大小。 maximumPoolSize是執行緒池的最大上限。 keepAliveTime是執行緒的存活時間,當執行緒池內的執行緒數大於corePoolSize,超出存活時間的空閒執行緒就會被回收。 unit就不用說了,剩下的三個參數看後文的分析。

預設的客製化執行緒池

ThreadPoolExecutor預設了一些已經客製化的線程池,由Executors裡的工廠方法創建。以下分析newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool的建立參數。

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newFixedThreadPool的corePoolSize和maximumPoolSize都設定為傳入的固定數量,keepAliveTim設定為0。執行緒池建立後,執行緒數量將會固定不變,適合需要執行緒很穩定的場合。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

newSingleThreadExecutor是執行緒數量固定為1的newFixedThreadPool版本,保證池內的任務序列。注意到回傳的是FinalizableDelegatedExecutorService,來看看原始碼:

static class FinalizableDelegatedExecutorService
        extends DelegatedExecutorService {
        FinalizableDelegatedExecutorService(ExecutorService executor) {
            super(executor);
        }
        protected void finalize() {
            super.shutdown();
        }
    }

FinalizableDelegatedExecutorService繼承了DelegatedExecutorService,僅在gc時增加關閉執行緒池的操作,再來看看DelegatedExecutorService的來源碼:

 

 static class DelegatedExecutorService extends AbstractExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) { e.execute(command); }
        public void shutdown() { e.shutdown(); }
        public List<Runnable> shutdownNow() { return e.shutdownNow(); }
        public boolean isShutdown() { return e.isShutdown(); }
        public boolean isTerminated() { return e.isTerminated(); }
        //...
    }

程式碼很簡單,DelegatedExecutorService包裝了ExecutorService,使其只暴露出ExecutorService的方法,因此不能再配置執行緒池的參數。本來,線程池創建的參數是可以調整的,ThreadPoolExecutor提供了set方法。使用newSingleThreadExecutor目的是產生單執行緒串列的執行緒池,如果還能配置執行緒池大小,那就沒意思了。

Executors也提供了unconfigurableExecutorService方法,將普通執行緒池包裝成不可設定的執行緒池。如果不想線程池被不明所以的後人修改,可以呼叫這個方法。

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

newCachedThreadPool產生一個會快取的執行緒池,執行緒數量可以從0到Integer.MAX_VALUE,逾時時間為1分鐘。線程池用起來的效果是:如果有空閒線程,會重複使用線程;如果沒有空閒線程,會新線程;如果線程空閒超過1分鐘,將會被回收。

newScheduledThreadPool

newScheduledThreadPool將會建立一個可定時執行任務的執行緒池。這個不打算在本文展開,後續會另開文章細講。

#

等待队列

newCachedThreadPool的线程上限几乎等同于无限,但系统资源是有限的,任务的处理速度总有可能比不上任务的提交速度。因此,可以为ThreadPoolExecutor提供一个阻塞队列来保存因线程不足而等待的Runnable任务,这就是BlockingQueue。

JDK为BlockingQueue提供了几种实现方式,常用的有:

ArrayBlockingQueue:数组结构的阻塞队列

LinkedBlockingQueue:链表结构的阻塞队列

PriorityBlockingQueue:有优先级的阻塞队列

SynchronousQueue:不会存储元素的阻塞队列

newFixedThreadPool和newSingleThreadExecutor在默认情况下使用一个无界的LinkedBlockingQueue。要注意的是,如果任务一直提交,但线程池又不能及时处理,等待队列将会无限制地加长,系统资源总会有消耗殆尽的一刻。所以,推荐使用有界的等待队列,避免资源耗尽。但解决一个问题,又会带来新问题:队列填满之后,再来新任务,这个时候怎么办?后文会介绍如何处理队列饱和。

newCachedThreadPool使用的SynchronousQueue十分有趣,看名称是个队列,但它却不能存储元素。要将一个任务放进队列,必须有另一个线程去接收这个任务,一个进就有一个出,队列不会存储任何东西。因此,SynchronousQueue是一种移交机制,不能算是队列。newCachedThreadPool生成的是一个没有上限的线程池,理论上提交多少任务都可以,使用SynchronousQueue作为等待队列正合适。

饱和策略

当有界的等待队列满了之后,就需要用到饱和策略去处理,ThreadPoolExecutor的饱和策略通过传入RejectedExecutionHandler来实现。如果没有为构造函数传入,将会使用默认的defaultHandler。

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
public static class AbortPolicy implements RejectedExecutionHandler {
       public AbortPolicy() { }
       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
           throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
       }
   }

AbortPolicy是默认的实现,直接抛出一个RejectedExecutionException异常,让调用者自己处理。除此之外,还有几种饱和策略,来看一下:

 

public static class DiscardPolicy implements RejectedExecutionHandler {
       public DiscardPolicy() { }
       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
       }
   }

DiscardPolicy的rejectedExecution直接是空方法,什么也不干。如果队列满了,后续的任务都抛弃掉。

 

 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
       public DiscardOldestPolicy() { }
       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
           if (!e.isShutdown()) {
               e.getQueue().poll();
               e.execute(r);
           }
       }
   }

DiscardOldestPolicy会将等待队列里最旧的任务踢走,让新任务得以执行。

    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        public CallerRunsPolicy() { }
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

最后一种饱和策略是CallerRunsPolicy,它既不抛弃新任务,也不抛弃旧任务,而是直接在当前线程运行这个任务。当前线程一般就是主线程啊,让主线程运行任务,说不定就阻塞了。如果不是想清楚了整套方案,还是少用这种策略为妙。

ThreadFactory

每当线程池需要创建一个新线程,都是通过线程工厂获取。如果不为ThreadPoolExecutor设定一个线程工厂,就会使用默认的defaultThreadFactory:

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}
static class DefaultThreadFactory implements ThreadFactory {
       private static final AtomicInteger poolNumber = new AtomicInteger(1);
       private final ThreadGroup group;
       private final AtomicInteger threadNumber = new AtomicInteger(1);
       private final String namePrefix;
       DefaultThreadFactory() {
           SecurityManager s = System.getSecurityManager();
           group = (s != null) ? s.getThreadGroup() :
                                 Thread.currentThread().getThreadGroup();
           namePrefix = "pool-" +
                         poolNumber.getAndIncrement() +
                        "-thread-";
       }
       public Thread newThread(Runnable r) {
           Thread t = new Thread(group, r,
                                 namePrefix + threadNumber.getAndIncrement(),
                                 0);
           if (t.isDaemon())
               t.setDaemon(false);
           if (t.getPriority() != Thread.NORM_PRIORITY)
               t.setPriority(Thread.NORM_PRIORITY);
           return t;
       }
   }

平时打印线程池里线程的name时,会输出形如pool-1-thread-1之类的名称,就是在这里设置的。这个默认的线程工厂,创建的线程是普通的非守护线程,如果需要定制,实现ThreadFactory后传给ThreadPoolExecutor即可。

不看代码不总结不会知道,光是线程池的创建就可以引出很多学问。别看平时创建线程池是一句代码的事,其实ThreadPoolExecutor提供了很灵活的定制方法。

欢迎留言和转发,下一篇打算分析线程池如何执行任务。

 以上就是Java 线程池的创建过程分析 的内容,更多相关内容请关注PHP中文网(www.php.cn)!

陳述:
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn