Java线程池框架源码分析(3)

这三个是ThreadPoolExecutor的成员变量,其中workQueue跟县城池没有关系。workQueue是一个线程安全的阻塞队列。

corePoolSize是线程池的核心大小,maximumPoolSize是线程池的最大大小。

当提交新任务时,如果ThreadPoolExecutor中有线程在运行,并且线程的数量小于corePoolSize,那么就会有新的线程被创建。如果当前运行的线程数大于corePoolSize,就会放到缓存队列workQueue中。如果缓冲队列也满了,就继续创建线程,直到线程的数量达到maximumPoolSize

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();

//判断如果当前运行的线程数小于 corePoolSize,添加新的线程(addWorker会添加一个新的线程,上面有介绍),方法直接返回。
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }


        //如果当前的运行的线程数大于或等于corePoolSize则新的任务会放到缓存队列中。
        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);

}

还有些其他的参数:

private volatile ThreadFactory threadFactory //线程的工厂函数。
 private volatile RejectedExecutionHandler handler;//任务拒绝的处理类。
 private volatile long keepAliveTime;//任务等待的是将。

ThreadPoolExecutor有几个构造方法来初始化这些参数。Executors类将这些参数简化了来获得一个ExecutorService的引用。

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

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

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

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

这四个方法中前两个的核心线程数和最大线程数相同,所有可运行的线程数是固定的,<=nThreads。当任务数大于nThreads时,就是放入缓冲队列中。  后两个方法中,线程数是无边界的,核心线程数是0,最大线程数是整型的最大值,然后如果有线程60秒内没有任务运行的话就销毁。每次有新的任务来,都会创建新的线程或使用以前创建的线程(60秒内没有任务运行的线程)。你可能有疑问,既然核心线程数是0,那么所有的任务不是都放到队里里了吗?那么现在就来看看SynchronousQueue这个队里,可以看看这里的介绍

回过头来看看任务提交方法的源码:

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {//这里是在往队列里方任务,如果不成功就会添加Worker(封装了线程对象)
            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);
    }

上面链接里提到:offer()往queue里放一个element后立即返回,如果碰巧这个element被另一个thread取走了,offer方法返回true,认为offer成功;否则返回false。

试想一下,第一次提交任务的时候,核心线程数为0,此时没有线程所以没有线程从workQueue中取东西,所以这里的workQueue.offer(command)会返回false,那么就会通过addWorker(command, false)创建一个新的线程。

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/e281b0e157df4a3c7317cb6013e4eab7.html