首頁 > Java > java教程 > java並發線程池:詳解ThreadPoolExecutor

java並發線程池:詳解ThreadPoolExecutor

php是最好的语言
發布: 2018-08-08 11:23:14
原創
1905 人瀏覽過

總結:線程池的特點是,在線程的數量=corePoolSize後,僅任務隊列滿了之後,才會從任務隊列中取出一個任務,然後構造一個新的線程,循環往復直到線程數量達到maximumPoolSize執行拒絕策略。

線程池-intsmaze

線程池的想法是:在系統中開闢一塊區域,其中存放一些待命的線程,這個區域被稱為線程池。如果有需要執行的任務,則從執行緒池中藉一個待命的執行緒來執行指定的任務,到任務結束可以再將所借執行緒歸還。這樣就避免了大量重複創建線程對象,浪費CPU,記憶體資源。

自訂執行緒池-intsmaze

如果觀察jdk提供的各種執行緒池的原始碼實作可以發現,除了jdk8新增的執行緒池newWorkStealingPool以外,都是基於對ThreadPoolExecutor的封裝實現,所以先來講解ThreadPoolExecutor的具體功能。

ThreadPoolExecutor詳解-intsmaze

#
ThreadPoolExecutor( corePoolSize,  maximumPoolSize,  keepAliveTime, TimeUnit unit, 
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
登入後複製

corePoolSize:指定執行緒池中執行緒數量

maximumPoolSize:最大執行緒數量

keepAliveTime:當線程數量超過corePoolSize時,多於的空閒線程的存活時間(超過這段時間,該空閒線程會被銷毀)。

unit:keepAliveTime的時間單位

workQueue:任務佇列,提交但是未執行的任務

threadFactory:建立執行緒的執行緒工廠,預設即可

handler:拒絕策略,當任務太多來不及處理,如何拒絕任務,預設為new AbortPolicy()策略。

        ExecutorService es = new ThreadPoolExecutor(3, 8, 60L,
                TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
                Executors.defaultThreadFactory(),                new RejectedExecutionHandler() {                    public void rejectedExecution(Runnable r,
                            ThreadPoolExecutor executor) {
                        System.out.println("discard");
                    }
                });
登入後複製

任務佇列--存放runnable物件-intsmaze

總結:執行緒池的特點是,在執行緒的數量= corePoolSize後,僅任務佇列滿了之後,才會從任務佇列中取出一個任務,然後建構一個新的線程,循環往復直到線程數量達到maximumPoolSize執行拒絕策略。

只要佇列實作BlockingQueue介面即可,注意ConcurrentLinkedQueue實作的最頂層的queue介面所以不能用在這裡。

常用的有如下:

SynchronousQueue:直接提交佇列,該佇列沒有容量,每一個插入操作都要等待一個對應的刪除操作,反之,每一個刪除操作都要等待對應的插入操作。所以他不保存任務,總是將任務提交給線程執行,如果沒有空閒的線程,則創建新的線程,當線程數量達到最大,則執行拒絕策略。

ArrayBlockingQueue:有界任務佇列,執行緒池的執行緒數小於corePoolSize,則建立新的線程,大於corePoolSize,則將新的任務加入等待隊列。若等待佇列已滿,則在總執行緒不大於maximumPoolSize下,建立新的執行緒執行任務,大於maximumPoolSize則執行拒絕策略。

LinkedBlockingQueue:無界佇列,除非系統資源耗盡,否則不存在任務入隊失敗的情況。線程池的線程數小於corePoolSize,則建立新的線程,大於corePoolSize,則將新的任務加入等待佇列。

PriortyBlockingQueue:優先任務佇列,可以控制任務的執行先後順序,是無界佇列。 ArrayBlockingQueue,LinkedBlockingQueue都是依照先進先出演算法處理任務的,PriorityBlockingQueue可以依照任務本身的優先順序先後執行。

拒絕策略-intsmaze

執行緒池中的執行緒用完了,同時等待佇列中的任務已經塞滿了,再也塞不下新任務了,就需要拒絕策略:處理任務數量超過系統實際承受能力時,處理方式。

jdk內建四種拒絕策略:

AbortPolicy:直接拋出例外(預設策略),就算執行緒池有空閒了,後面的執行緒也無法在執行了,要想後面的執行緒可以運行,要捕捉異常訊息。

CallerRunsPolicy:此策略直接在呼叫者執行緒中執行目前被丟棄的任務。顯然這樣做不會真的丟棄任務,但是任務提交線程的效能極有可能會急劇下降。

DiscardOldestPolicy:將丟棄最老的一個請求,也就是即將被執行的一個任務,並嘗試再次提交目前任務。

DiscardPolicy:默默丟棄無法處理的任務,不予任何處理。如果允許任務遺失,這可能是最好的解決方案。在線程池不空閒的時候,提交的任務都會丟棄,當有空閒的線程時提交的任務會執行。

下面是jdk的拒絕策略來源碼-intsmaze

   public static class CallerRunsPolicy implements RejectedExecutionHandler {        public CallerRunsPolicy() { }        /**
         * 直接在调用者线程中运行当前被丢弃的任务         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {            if (!e.isShutdown()) {
                r.run();
            }
        }
    }    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());
        }
    }    public static class DiscardPolicy implements RejectedExecutionHandler {        public DiscardPolicy() { }        /**
         * Does nothing, which has the effect of discarding task r.         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }    public static class DiscardOldestPolicy implements RejectedExecutionHandler {        public DiscardOldestPolicy() { }        /**
         * 将丢弃最老的一个请求,也就是即将被执行的一个任务,并尝试再次提交当前任务。         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }
登入後複製

#

总结:AbortPolicy策略下,我们要catch异常,这样我们可以捕获到哪些任务被丢弃了。如果采用其他的策略,丢弃的任务无法定位的,只能通过下列程序中es.submit(new MyTask(i));任务之前打印该任务,运行任务的run()逻辑是,在打印任务信息,两处日志比对来定位哪些任务被丢弃了。

public class MyTask implements Runnable
{    private int number;    
    public MyTask(int number) {        super();        this.number = number;
    }    public void run() {
        System.out.println(System.currentTimeMillis()+"thread id:"+Thread.currentThread().getId()+"==="+number);        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}    public static void main(String[] args)  {//        ExecutorService es=new ThreadPoolExecutor(5,5,60L, TimeUnit.SECONDS, 
//                new ArrayBlockingQueue<Runnable>(1), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());        
//        ExecutorService es=new ThreadPoolExecutor(5,5,60L, TimeUnit.SECONDS,//                new ArrayBlockingQueue<Runnable>(5), Executors.defaultThreadFactory(),new ThreadPoolExecutor.CallerRunsPolicy());        
//        ExecutorService es=new ThreadPoolExecutor(5,5,60L, TimeUnit.SECONDS,//                new ArrayBlockingQueue<Runnable>(5), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());        
        ExecutorService es=new ThreadPoolExecutor(5,5,60L, TimeUnit.SECONDS,                new ArrayBlockingQueue<Runnable>(5), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardOldestPolicy());        for(int i=0;i<10000;i++)
        {            try {
                System.out.println(i);
                es.submit(new MyTask(i));
                Thread.sleep(100);
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("------------------------"+i);
            }
        }
    }
登入後複製

线程池执行逻辑源码解析-intsmaze

      public Future<?> submit(Runnable task) {        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);        return ftask;
    }    
       /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null     */
    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&#39;t, by returning false.
         *如果少于corePoolSize线程正在运行,首先尝试用给定的命令启动一个新的线程任务。 
         自动调用addWorker检查runState和workerCount,
         
         
         * 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);
    }    
    final void reject(Runnable command) {
        handler.rejectedExecution(command, this);//这里就是调用我们传入的拒绝策略对象的方法    }    
     /**
     * Dispatch an uncaught exception to the handler. This method is
     * intended to be called only by the JVM.     */
    private void dispatchUncaughtException(Throwable e) {
        getUncaughtExceptionHandler().uncaughtException(this, e);
    }
登入後複製

jdk的线程池实现类-intsmaze

newFixedThreadPoo-intsmaze

任务队列为LinkedBlockingQueue中(长度无限),线程数量和最大线程数量相同。功能参考前面的任务队列总结。

ExecutorService es=Executors.newFixedThreadPool(5);//参数同时指定线程池中线程数量为5,最大线程数量为5public static ExecutorService newFixedThreadPool(int nThreads) {    return new ThreadPoolExecutor(nThreads, nThreads,                                  0L, TimeUnit.MILLISECONDS,                                  new LinkedBlockingQueue<Runnable>());
}
登入後複製

newSingleThreadExecutor-intsmaze

任务队列LinkedBlockingQueue中(长度无限),线程数量和最大线程数量均为1。

ExecutorService es=Executors.newSingleThreadExecutor();//线程池中线程数量和最大线程数量均为1.public static ExecutorService newSingleThreadExecutor() {    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,                                0L, TimeUnit.MILLISECONDS,                                new LinkedBlockingQueue<Runnable>()));
}
登入後複製

newCachedThreadPool-intsmaze

任务队列为SynchronousQueue,线程数量为0,最大线程数量为Integer.MAX_VALUE,所以只要有任务没有空闲线程就会创建就新线程。

ExecutorService es=Executors.newCachedThreadPool();//指定线程池中线程数量为0,最大线程数量为Integer.MAX_VALUE,任务队列为SynchronousQueuepublic static ExecutorService newCachedThreadPool() {    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,                                  60L, TimeUnit.SECONDS,                                  new SynchronousQueue<Runnable>());
}
登入後複製

newScheduledThreadPool- -定时线程-intsmaze

任务队列为new DelayedWorkQueue(),返回的对象在ExecutorService接口上扩展了在指定时间执行某认为的功能,在某个固定的延时之后执行或周期性执行某个任务。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {        return new ScheduledThreadPoolExecutor(corePoolSize);
}public ScheduledThreadPoolExecutor(int corePoolSize) {        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,              new DelayedWorkQueue());
}public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
}
登入後複製

newSingleThreadScheduledExecutor- -定时线程-intsmaze

相当于newScheduledThreadPool(int corePoolSize)corePoolSize设置为1。

ScheduledExecutorService es=Executors.newSingleThreadScheduledExecutor();
登入後複製

延迟线程池

class MyScheduledTask implements Runnable
{ private String tname; public MyScheduledTask(String tname)
 {  this.tname=tname;
 } public void run()
 {
  System.out.println(tname+"任务时延2秒执行!!!");
 }
}public class intsmaze
{ public static void main(String[] args)
 {  ScheduledExecutorService scheduledThreadPool                       =Executors.newScheduledThreadPool(2);
  MyScheduledTask mt1=new MyScheduledTask("MT1");
  scheduledThreadPool.schedule(mt1,2,TimeUnit.SECONDS);
 }
}
登入後複製

newWorkStealingPool java8新增连接池-intsmaze

    public static ExecutorService newWorkStealingPool(int parallelism) {        return new ForkJoinPool
            (parallelism,
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,             null, true);
    }//创建指定数量的线程池来执行给定的并行级别,还会使用多个队列减少竞争
    public static ExecutorService newWorkStealingPool() {        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,             null, true);
    }//前一个方法的简化,如果当前机器有4个CPU,则目标的并行级别被设置为4。
登入後複製

关闭线程池(很少使用,除了切换数据源时需要控制)-intsmaze

希望程序执行完所有任务后退出,调用ExecutorService接口中的shutdown(),shutdownNow()方法。

用完一个线程池后,应该调用该线程池的shutdown方法,将启动线程池的关闭序列。调用shutdown方法后,线程池不在接收新的任务,但是会将以前所有已经提交的任务执行完。当线程池中的所有任务都执行完后,线程池中的所有线程都会死亡;shutdownNow方法会试图停止所有正在执行的活动任务,暂停处理正在等待的任务,并返回等待执行的任务列表。

线程池优化-intsmaze

一般来说确定线程池的大小需要考虑CPU数量,内存大小,JDBC连接等因素。在《java并发编程实践》一书中给出了一个估算线程池大小的经验公式:

Ncpu=CPU的数量

Ucpu=目标CPU的使用率,0<=Ucpu<=1

W/C=等待时间与计算时间的比率

为保持处理器达到期望的使用率,最优的线程池的大小等于:

Nthreads=Ncpu*Ucpu*(1+W/C)

在java中,可以通过

Runtime.getRuntime().availableProcessors()
登入後複製

取得可以CPU数量。

相关推荐:

Java中线程池的图文代码详解

ThreadPoolExecutor线程池之submit方法

JAVA中ThreadPoolExecutor线程池的submit方法详解

以上是java並發線程池:詳解ThreadPoolExecutor的詳細內容。更多資訊請關注PHP中文網其他相關文章!

來源:php.cn
本網站聲明
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
熱門教學
更多>
最新下載
更多>
網站特效
網站源碼
網站素材
前端模板