`
Donald_Draper
  • 浏览: 979840 次
社区版块
存档分类
最新评论

ThreadPoolExecutor解析二(线程工厂、工作线程,拒绝策略等)

    博客分类:
  • JUC
阅读更多
Executor接口的定义:http://donald-draper.iteye.com/blog/2365625
ExecutorService接口定义:http://donald-draper.iteye.com/blog/2365738
Future接口定义:http://donald-draper.iteye.com/blog/2365798
FutureTask解析:http://donald-draper.iteye.com/blog/2365980
CompletionService接口定义:http://donald-draper.iteye.com/blog/2366239
ExecutorCompletionService解析:http://donald-draper.iteye.com/blog/2366254
AbstractExecutorService解析:http://donald-draper.iteye.com/blog/2366348
ScheduledExecutorService接口定义:http://donald-draper.iteye.com/blog/2366436
ThreadPoolExecutor解析一(核心线程池数量、线程池状态等) :
http://donald-draper.iteye.com/blog/2366934

上一篇文章我们讲到线程池ThreadPoolExecutor的java doc使用说明和变量的定义,先回顾一下:
   ThreadPoolExecutor的变量主要有核心线程池数量corePoolSize和最大线程池数量maximumPoolSize,即在当前任务线程数大于核心线程数量时,是否(allowCoreThreadTimeOut)允许空闲任务线程等,保活keepAliveTime时间,等待新任务的到来。一个线程工厂ThreadFactory用于创建任务线程,一个拒绝任务处理器RejectedExecutionHandler,默认的拒绝任务策略为AbortPolicy,抛出运行时异常,当然还有直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,还有调用者执行任务策略CallerRunsPolicy。上面的变量为volatile,以便线程池执行操作时,可以使用最新的变量。
    一个阻塞的任务队列final BlockingQueue<Runnable> workQueue,阻塞队列可以为Linked,Array,Delay,SynchronousQueue等阻塞类型,具体可以根据场景选择。默认为LinkedBlockingQueue队列,一般判断队列是否为空,用isEmpty方法,LinkedBlockingQueue一般用于任务相互之间独立,没有交叉,可独立执行。如果用SynchronousQueue,则可用poll方法判断,同步队列一般用于任务之间有依赖的关系的场景,一个任务执行依赖于另一个任务的结果。DelayQueue队列用于定时任务。ArrayBlockingQueue队列一般可以用于
资源有限情况,可以避免资源被耗尽。一个AtomicInteger的ctl用于包装线程状态runState和任务线程数workerCount;低29位保存任务线程数,高两位用于存储线程池状态,线程池状态已用有四种RUNNING,SHUTDOWN ,STOP,TIDYING ,TERMINATED。
RUNNING:接受新的任务,处理队列任务;
SHUTDOWN:不在接受新的任务,处理队列任务;
STOP:不在接受新任务,不处理队列任务,中断正在执行的任务线程;
TIDYING:所有的任务已经结束,任务线程为0,线程转换到TIDYING;
TERMINATED:线程池已结束,即terminated()方法执行完。
线程的状态是可以数字化比较的。

    一个任务线程集final HashSet<Worker> workers,largestPoolSize记录线程池的最大任务线程数,completedTaskCount为完成任务计数器,在任务线程结束时,更新。一个可重入锁mainLock,用于保护非线程安全的变量如workers,largestPoolSize,completedTaskCount。
一个等待线程池结束条件termination,用于控制超时等待线程池关闭。



先来看ThreadPoolExecutor的几种构造方法
 // Public constructors and methods

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default thread factory and rejected execution handler.
     * It may be more convenient to use one of the {@link Executors} factory
     * methods instead of this general purpose constructor.
     *
     根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列构造ThreadPoolExecutor,
     线程池工厂默认为Executors.defaultThreadFactory(),拒绝策略为默认的AbortPolicy。
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default rejected execution handler.
     *
     根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,线程池工厂构造ThreadPoolExecutor,
     拒绝策略为默认的AbortPolicy。
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default thread factory.
     *
     根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,拒绝策略构造ThreadPoolExecutor,
     线程池工厂默认为Executors.defaultThreadFactory()。
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,线程池工厂,拒绝策略构造ThreadPoolExecutor。
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

有了上面的构造ThreadPoolExecutor,我们来看一下默认的线程工厂和拒绝策略。
先看拒绝策略:
//ThreadPoolExecutor
 /**
     * A handler for rejected tasks that throws a
     默认的线程池拒绝任务处理策略
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
	 拒绝策略为直接抛出RejectedExecutionException
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always.
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

AbortPolicy其实为ThreadPoolExecutor一个静态内部类。
我们既然看到拒绝策略把其他几个也一起看一下:
直接丢弃策略DiscardPolicy:
/**
     * A handler for rejected tasks that silently discards the
     * rejected task.
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         *
	 不做任何工作直接丢弃任务
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

丢弃旧的任务DiscardOldestPolicy:
/**
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries {@code execute}, unless the executor
     * is shut down, in which case the task is discarded.
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         获取并忽略执行器下一个将会执行的任务,执行新提交的任务,如果有任务线程立刻可用,
	 则重新尝试执行任务r,如果线程池已关闭,则丢弃任务。
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
	        //如果线程中没关闭,任务队列头任务出队列,并不执行,执行的是刚提交的任务 r
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

调用者执行任务CallerRunsPolicy:
/* Predefined RejectedExecutionHandlers */

    /**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         调用者执行任务,如果线程池已关闭,则丢弃任务
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
	        //如果线程池没关闭,则直接执行任务
                r.run();
            }
        }
    }

上面所讲的四种丢弃策略都是ThreadPoolExecutor的内部类,分别为默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。
DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。

再看默认的线程工厂
//Executors
Executors.defaultThreadFactory()。


  
 /**
     * Returns a default thread factory used to create new threads.
     * This factory creates all new threads used by an Executor in the
     * same {@link ThreadGroup}. If there is a {@link
     * java.lang.SecurityManager}, it uses the group of {@link
     * System#getSecurityManager}, else the group of the thread
     * invoking this <tt>defaultThreadFactory</tt> method. Each new
     * thread is created as a non-daemon thread with priority set to
     * the smaller of <tt>Thread.NORM_PRIORITY</tt> and the maximum
     * priority permitted in the thread group.  New threads have names
     * accessible via {@link Thread#getName} of
     * [i]pool-N-thread-M[/i], where [i]N[/i] is the sequence
     * number of this factory, and [i]M[/i] is the sequence number
     * of the thread created by this factory.
     返回一个默认的线程工厂用于创建线程,工厂创建分组相同的线程,交由执行器执行。
     如果有java.lang.SecurityManager,则用System#getSecurityManager线程组,
     否则用调用者的线程组。创建的新线程为非守护模式,优先级在 MIN_PRIORITY和MAX_PRIORITY之间,
     默认为NORM_PRIORITY。可以通过Thread#getName获取线程name,默认为pool-N-thread-M,
     N为线程池编号,M为线程编号。
     * @return a thread factory
     */
    public static ThreadFactory defaultThreadFactory() {
        return new DefaultThreadFactory();
    }



//Executors
  
  /**
     * The default thread factory,默认的线程工厂
     */
    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();
	   //构造一个线程池工厂,线程池编号自增1
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }
        //创建新线程,线程名为pool-1-thread-1, 1为当前线程工程的线程
        public Thread newThread(Runnable r) {
	   //根据分组,Runnable,名字创建线程,线程编号自增1
            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;
        }
    }

DefaultThreadFactory为Executors的内部类,
//Thread
public final static int MIN_PRIORITY = 1;//最小优先级
public final static int NORM_PRIORITY = 5;//默认
public final static int MAX_PRIORITY = 10;//最大优先级



再来看执行任务的工作线程Worker:
/**
     * Class Worker mainly maintains interrupt control state for
     * threads running tasks, along with other minor bookkeeping.
     * This class opportunistically extends AbstractQueuedSynchronizer
     * to simplify acquiring and releasing a lock surrounding each
     * task execution.  This protects against interrupts that are
     * intended to wake up a worker thread waiting for a task from
     * instead interrupting a task being run.  We implement a simple
     * non-reentrant mutual exclusion lock rather than use
     * ReentrantLock because we do not want worker tasks to be able to
     * reacquire the lock when they invoke pool control methods like
     * setCorePoolSize.  Additionally, to suppress interrupts until
     * the thread actually starts running tasks, we initialize lock
     * state to a negative value, and clear it upon start (in
     * runWorker).
     Worker主要为任务线程维护中断控制状态和其他次要状态记录。 Worker简单实现了
     AQS在任务线程执行前lock,任务执行完unlock。加锁的目的主要是保护任务线程的执行,
     线程池唤醒一个任务线程等待任务,而不是中断当前正在执行任务的线程去执行任务。我们使用了一个
     非重入互质锁,而不是ReentrantLock,这样做的目的是以防在任务执行的过程,线程池控制
     方法的改变,对任务线程执行的影响,比如setCorePoolSize方法。另外为了防止任务线程在实际
     执行前被中断,我们初始化锁状态为-1,在runWorker方法中,我们会清除它。
     */
    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;//任务线程
        /** Initial task to run.  Possibly null. */
        Runnable firstTask; //任务
        /** Per-thread task counter */
        volatile long completedTasks; //线程完成的任务计数

        /**
         * Creates with given first task and thread from ThreadFactory.
	 根据给定的任务,用线程工厂创建任务线程
         * @param firstTask the first task (null if none)
         *
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        /** Delegates main run loop to outer runWorker  */
        public void run() {
	    //运行线程
            runWorker(this);
        }

        // Lock methods
        //
        // The value 0 represents the unlocked state. 开锁状态
        // The value 1 represents the locked state.闭锁状态

        protected boolean isHeldExclusively() {
            return getState() != 0;//是否持有锁
        }
       //尝试获取锁
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
       //尝试释放锁
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock()        { acquire(1); }//加锁
        public boolean tryLock()  { return tryAcquire(1); }//尝试加锁
        public void unlock()      { release(1); }//解锁
        public boolean isLocked() { return isHeldExclusively(); }//是否锁住
        //如果任务正在执行,则中断
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
		    //如果锁处于关闭状态,且任务线程不为null,非出于中断状态,则中断当前线程
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

来看Worker的run方法:

 /** Delegates main run loop to outer runWorker  */
        public void run() {
	    //运行线程
            runWorker(this);
	}

//ThreadPoolExecutor
/**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     *
     工作线程运行的主要方法。重复从任务队列获取任务,并执行,主要有一下几点注意事项:
     * 1. We may start out with an initial task, in which case we
     * don't need to get the first one. Otherwise, as long as pool is
     * running, we get tasks from getTask. If it returns null then the
     * worker exits due to changed pool state or configuration
     * parameters.  Other exits result from exception throws in
     * external code, in which case completedAbruptly holds, which
     * usually leads processWorkerExit to replace this thread.
     *
     1.我们也许在开始时有一个初始化的任务,这是不需要从任务队列获取任务。
     只要线程池在运行,则从getTask方法获取任务,由于线程状态或配置参数的
     改变,获取任务为null时,则工作线程退出。如果执行过程中抛出异常,则
     调用processWorkerExit方法,取代当前线程
     * 2. Before running any task, the lock is acquired to prevent
     * other pool interrupts while the task is executing, and
     * clearInterruptsForTaskRun called to ensure that unless pool is
     * stopping, this thread does not have its interrupt set.
     *
     2.在执行任务前,获取lock,防止任务在执行的过程中被其他线程池中断。
     clearInterruptsForTaskRun清除正在运行任务的中断位,在线程没有关闭的情况下。
     * 3. Each task run is preceded by a call to beforeExecute, which
     * might throw an exception, in which case we cause thread to die
     * (breaking loop with completedAbruptly true) without processing
     * the task.
     *
     3.在任务执行前,调用beforeExecute,这个可能会抛出异常,将导致工作线程
     没有处理任务的情况下死掉。
     * 4. Assuming beforeExecute completes normally, we run the task,
     * gathering any of its thrown exceptions to send to
     * afterExecute. We separately handle RuntimeException, Error
     * (both of which the specs guarantee that we trap) and arbitrary
     * Throwables.  Because we cannot rethrow Throwables within
     * Runnable.run, we wrap them within Errors on the way out (to the
     * thread's UncaughtExceptionHandler).  Any thrown exception also
     * conservatively causes thread to die.
     *
     4.如果beforeExecute正常完成,执行任务,收集运行中的异常,交由afterExecute处理。
     我们独立地处理运行异常,错误和任意异常。由于我们不能在Runnable.run中重新抛异常,
     我们可以将异常包装在UncaughtExceptionHandler中。任何异常可能导致线程死掉
     * 5. After task.run completes, we call afterExecute, which may
     * also throw an exception, which will also cause thread to
     * die. According to JLS Sec 14.20, this exception is the one that
     * will be in effect even if task.run throws.
     *
     5.在任务执行完,我们可以调用afterExecute,方法可能抛出异常导致工作线程死掉。
     根据JLS,Java语言规范,即使这个异常有执行任务抛出,已有可能导致线程死掉。
     * The net effect of the exception mechanics is that afterExecute
     * and the thread's UncaughtExceptionHandler have as accurate
     * information as we can provide about any problems encountered by
     * user code.
     afterExecute方法和线程的UncaughtExceptionHandler,给我们提供了针对错误的便利。
     *
     * @param w the worker
     */
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();//当前线程
        Runnable task = w.firstTask;//工作线程任务
        w.firstTask = null;
	//任务线程的锁状态默认为-1,此时解锁+1,变为0,即锁打开状态,允许中断,在任务未执行之前,不允许中断。
        w.unlock(); // allow interrupts,
        boolean completedAbruptly = true;//完成后是否可以中断
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted; 
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
		//如果线程池正在Stop,则确保线程中断;
		//如果非处于Stop之后的状态,则判断是否中断,如果中断则判断线程池是否已关闭
		//如果线程池正在关闭,但没有中断,则中断线程池
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
		   //执行前工作
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
		        //执行任务
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
		       //执行后工作
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
		    //任务线程完成任务数量加1,释放锁
                    w.completedTasks++;
                    w.unlock();
                }
            }
	    //任务已执行完不可以中断
            completedAbruptly = false;
        } finally {
	    //处理任务完成后的工作
            processWorkerExit(w, completedAbruptly);
        }
    }

runWorker有几点要关注:
1. //执行前工作
beforeExecute(wt, task);

2.//执行后工作
afterExecute(task, thrown);

3.
//处理任务完成后的工作
processWorkerExit(w, completedAbruptly);

分别来看每一点:

1. //执行前工作
beforeExecute(wt, task);
    /**
     * Method invoked prior to executing the given Runnable in the
     * given thread.  This method is invoked by thread {@code t} that
     * will execute task {@code r}, and may be used to re-initialize
     * ThreadLocals, or to perform logging.
     *
     * <p>This implementation does nothing, but may be customized in
     * subclasses. Note: To properly nest multiple overridings, subclasses
     * should generally invoke {@code super.beforeExecute} at the end of
     * this method.
     *
     待子类扩展
     * @param t the thread that will run task {@code r}
     * @param r the task that will be executed
     */
    protected void beforeExecute(Thread t, Runnable r) { }

2.//执行后工作
afterExecute(task, thrown);

/**
     * Method invoked upon completion of execution of the given Runnable.
     * This method is invoked by the thread that executed the task. If
     * non-null, the Throwable is the uncaught {@code RuntimeException}
     * or {@code Error} that caused execution to terminate abruptly.
     *
    处理任务完成工作,和在任务执行过程中,未捕捉的异常等
     * <p>This implementation does nothing, but may be customized in
     * subclasses. Note: To properly nest multiple overridings, subclasses
     * should generally invoke {@code super.afterExecute} at the
     * beginning of this method.
     *
     * <p><b>Note:</b> When actions are enclosed in tasks (such as
     * {@link FutureTask}) either explicitly or via methods such as
     * {@code submit}, these task objects catch and maintain
     * computational exceptions, and so they do not cause abrupt
     * termination, and the internal exceptions are [i]not[/i]
     * passed to this method. If you would like to trap both kinds of
     * failures in this method, you can further probe for such cases,
     * as in this sample subclass that prints either the direct cause
     * or the underlying exception if a task has been aborted:
     *
     *  <pre> {@code
     * class ExtendedExecutor extends ThreadPoolExecutor {
     *   // ...
     *   protected void afterExecute(Runnable r, Throwable t) {
     *     super.afterExecute(r, t);
     *     if (t == null && r instanceof Future<?>) {
     *       try {
     *         Object result = ((Future<?>) r).get();
     *       } catch (CancellationException ce) {
     *           t = ce;
     *       } catch (ExecutionException ee) {
     *           t = ee.getCause();
     *       } catch (InterruptedException ie) {
     *           Thread.currentThread().interrupt(); // ignore/reset
     *       }
     *     }
     *     if (t != null)
     *       System.out.println(t);
     *   }
     * }}</pre>
     *
     待子类扩展
     * @param r the runnable that has completed
     * @param t the exception that caused termination, or null if
     * execution completed normally
     */
    protected void afterExecute(Runnable r, Throwable t) { }

3.
从现在开始我们把以前的任务线程一律改为工作线程,下面的工作线程及我们前面所说的任务线程。
//处理任务完成后的工作
processWorkerExit(w, completedAbruptly);


/**
     * Performs cleanup and bookkeeping for a dying worker. Called
     * only from worker threads. Unless completedAbruptly is set,
     * assumes that workerCount has already been adjusted to account
     * for exit.  This method removes thread from worker set, and
     * possibly terminates the pool or replaces the worker if either
     * it exited due to user task exception or if fewer than
     * corePoolSize workers are running or queue is non-empty but
     * there are no workers.
     *
     * @param w the worker
     * @param completedAbruptly if the worker died due to user exception
     */
    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
	    //如果任务线程被中断,则工作线程数量减1
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
	    //增加任务完成数量
            completedTaskCount += w.completedTasks;
	    //从工作线程集中,移除工作线程
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }
        //尝试结束线程池,这个我们后面再说
        //这个实在当前线程池正在关闭,但任务队列不为空,当前工作线程执行完任务,则尝试关闭空闲线程,更新线程池状态
        tryTerminate();

        int c = ctl.get();
	//如果线程处于运行中
        if (runStateLessThan(c, STOP)) {
	     //如果线程已完成
            if (!completedAbruptly) {
	        //获取当前核心线程数
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
		   //如果允许空闲工作线程等待任务,且任务队列不为空,则min为1
                    min = 1;
		//如果当前工作线程大于核心工作线程数,则返回
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

本想以后再看,现在看了吧
我们再来看看tryTerminate
 /**
     * Transitions to TERMINATED state if either (SHUTDOWN and pool
     * and queue empty) or (STOP and pool empty).  If otherwise
     * eligible to terminate but workerCount is nonzero, interrupts an
     * idle worker to ensure that shutdown signals propagate. This
     * method must be called following any action that might make
     * termination possible -- reducing worker count or removing tasks
     * from the queue during shutdown. The method is non-private to
     * allow access from ScheduledThreadPoolExecutor.
     */
    final void tryTerminate() {
        //自旋尝试关闭线程池
        for (;;) {
            int c = ctl.get();
	    //如果线程池正在运行,或关闭但队列不为空,则返回
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            if (workerCountOf(c) != 0) { // Eligible to terminate
	        //如果工作线程不为空,则中断空闲工作线程
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
	        //线程池已关闭,任务队列为空,工作线程为0,更新线程池状态为TIDYING
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
		        //执行结束工作
                        terminated();
                    } finally {
		        //线程池已结束
                        ctl.set(ctlOf(TERMINATED, 0));
			//唤醒等待线程池结束的线程
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }

来看尝试结束线程池的这一点,
//如果工作线程不为空,则中断空闲工作线程
interruptIdleWorkers(ONLY_ONE);
  /**
     * Interrupts threads that might be waiting for tasks (as
     * indicated by not being locked) so they can check for
     * termination or configuration changes. Ignores
     * SecurityExceptions (in which case some threads may remain
     * uninterrupted).
     *
     中断等待任务的空闲非锁住状态的工作线程
     * @param onlyOne If true, interrupt at most one worker. This is
     * called only from tryTerminate when termination is otherwise
     * enabled but there are still other workers.  In this case, at
     * most one waiting worker is interrupted to propagate shutdown
     * signals in case all threads are currently waiting.
     * Interrupting any arbitrary thread ensures that newly arriving
     * workers since shutdown began will also eventually exit.
     * To guarantee eventual termination, it suffices to always
     * interrupt only one idle worker, but shutdown() interrupts all
     * idle workers so that redundant workers exit promptly, not
     * waiting for a straggler task to finish.
     如果onlyOne为true以为着,只中断最多一个空闲工作线程,这个在关闭线程池时,
     调用或关闭的过程中,工作线程完成任务调用。
     */
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
	        //遍历工作线程集
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {//锁打开说明,工作线程空闲
                    try {
		        //如果工作线程非中断,且空闲,尝试获取锁,获取锁成功,则中断工作线程
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
		    //如果是只中断一个空闲线程,则结束本次中断空闲线程任务
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

从上可以看出尝试结束线程池所做的工作为,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。

总结:
   
     默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。
拒绝策略都为ThreadPoolExecutor的内部类。
   默认的线程工厂DefaultThreadFactory为Executors的内部类, 用于创建线程,工厂创建分组相同的线程,交由执行器执行。如果有java.lang.SecurityManager,则用System#getSecurityManager线程组,否则用调用者的线程组。创建的新线程为非守护模式,优先级在 MIN_PRIORITY和MAX_PRIORITY之间,默认为NORM_PRIORITY。可以通过Thread#getName获取线程name,默认为pool-N-thread-M,N为线程池编号,M为线程编号。
   Worker包装了任务线程,主要是为了维护中断控制状态和其他次要状态记录,及任务的执行。Worker同时继承了AQS,在任务线程执行前lock,任务执行完unlock。加锁的目的主要是保护任务线程的执行,线程池唤醒一个任务线程等待任务,而不是中断当前正在执行任务的线程去执行任务。Worker使用了一个 非重入互质锁,而不是ReentrantLock,这样做的目的是以防在任务执行的过程,线程池控制方法的改变,对任务线程执行的影响,比如setCorePoolSize方法。另外为了防止任务线程在实际执行前被中断,我们初始化锁状态为-1,在runWorker方法中,我们会清除它。runWorker执行任务时,首先释放锁,此时锁打开,允许中断,如果线程池正在stop,确保线程池已中断,否则
做执行前工作,执行任务,做执行后工作,如果任务被中断,则工作线程数量减1;
如果任务完成,则更新完成任务数量,从工作任务集中移除工作线程,尝试结束线程池。
    尝试结束线程池,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,
或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。


ThreadPoolExecutor解析三(线程池执行提交任务):
http://donald-draper.iteye.com/blog/2367199



//RejectedExecutionHandler
package java.util.concurrent;

/**
 * A handler for tasks that cannot be executed by a {@link ThreadPoolExecutor}.
 *
 * @since 1.5
 * @author Doug Lea
 */
public interface RejectedExecutionHandler {

    /**
     * Method that may be invoked by a {@link ThreadPoolExecutor} when
     * {@link ThreadPoolExecutor#execute execute} cannot accept a
     * task.  This may occur when no more threads or queue slots are
     * available because their bounds would be exceeded, or upon
     * shutdown of the Executor.
     *
     当线程池执行器不能接受任务时被调用。这种情况的发生可能由于没有任务线程可用,
     或任务队列已满,或执行器关闭。
     * <p>In the absence of other alternatives, the method may throw
     * an unchecked {@link RejectedExecutionException}, which will be
     * propagated to the caller of {@code execute}.
     *
     在一些实现策略中,可能抛出一个RejectedExecutionException,也可能有调用者去执行任务。
     * @param r the runnable task requested to be executed
     * @param executor the executor attempting to execute this task
     * @throws RejectedExecutionException if there is no remedy
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}



//ThreadFactory
package java.util.concurrent;

/**
 * An object that creates new threads on demand.  Using thread factories
 * removes hardwiring of calls to {@link Thread#Thread(Runnable) new Thread},
 * enabling applications to use special thread subclasses, priorities, etc.
 *
 根据需求创建线程。用线程工厂取代Thread#Thread(Runnable)常见新线程,使应用可以使用
 一些特殊的线程子类,比如具有优先级线程等
 * <p>
 * The simplest implementation of this interface is just:
 * <pre>
 * class SimpleThreadFactory implements ThreadFactory {
 *   public Thread newThread(Runnable r) {
 *     return new Thread(r);
 *   }
 * }
 * </pre>
 *
 * The {@link Executors#defaultThreadFactory} method provides a more
 * useful simple implementation, that sets the created thread context
 * to known values before returning it.
 * @since 1.5
 * @author Doug Lea
 */
public interface ThreadFactory {

    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     创建一个线程初始化线程的优先级,名字,是否为守候,线程组
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     *         create a thread is rejected
     */
    Thread newThread(Runnable r);
}

//Thread 线程未捕捉异常处理器
// Added in JSR-166

    /**
     * Interface for handlers invoked when a <tt>Thread</tt> abruptly
     * terminates due to an uncaught exception.
     * <p>When a thread is about to terminate due to an uncaught exception
     * the Java Virtual Machine will query the thread for its
     * <tt>UncaughtExceptionHandler</tt> using
     * {@link #getUncaughtExceptionHandler} and will invoke the handler's
     * <tt>uncaughtException</tt> method, passing the thread and the
     * exception as arguments.
     * If a thread has not had its <tt>UncaughtExceptionHandler</tt>
     * explicitly set, then its <tt>ThreadGroup</tt> object acts as its
     * <tt>UncaughtExceptionHandler</tt>. If the <tt>ThreadGroup</tt> object
     * has no
     * special requirements for dealing with the exception, it can forward
     * the invocation to the {@linkplain #getDefaultUncaughtExceptionHandler
     * default uncaught exception handler}.
     *
     * @see #setDefaultUncaughtExceptionHandler
     * @see #setUncaughtExceptionHandler
     * @see ThreadGroup#uncaughtException
     * @since 1.5
     */
    public interface UncaughtExceptionHandler {
        /**
         * Method invoked when the given thread terminates due to the
         * given uncaught exception.
         * <p>Any exception thrown by this method will be ignored by the
         * Java Virtual Machine.
         * @param t the thread
         * @param e the exception
         */
        void uncaughtException(Thread t, Throwable e);
    }

    // null unless explicitly set
    private volatile UncaughtExceptionHandler uncaughtExceptionHandler;

    // null unless explicitly set
    private static volatile UncaughtExceptionHandler defaultUncaughtExceptionHandler;

0
0
分享到:
评论

相关推荐

    ThreadPoolExecutor源码解析.pdf

    - **ReentrantLock mainLock**:用于在更新内部数据(如线程数量、运行状态、工作线程增减等)时进行同步。 - **Condition termination**:支持awaitTermination方法的等待条件。 - **Worker类**:线程池的核心...

    线程池原理-ThreadPoolExecutor源码解析

    线程池原理-ThreadPoolExecutor源码解析 1.构造方法及参数 2.阻塞对列: BlockingQueue 3.线程工厂: DefaultThreadFactory 4.拒绝策略: RejectedExecutionHandler 5.执行线程 Executor

    Android 线程 解析

    为了解决主线程的性能问题,开发者通常会在后台创建工作线程来执行耗时任务,例如网络请求、数据库操作、大量计算等。常见的工作线程实现方式有:子线程、IntentService、HandlerThread、AsyncTask以及使用线程池...

    Java Thread多线程全面解析

    除了上述创建线程的方式,Java还提供了Callable和Future接口,用于创建带返回值的线程,以及ExecutorService和ThreadPoolExecutor等高级线程管理工具,以更灵活地控制线程的执行和管理。 在实际开发中,理解和掌握...

    java线程、线程池、xml解析入门

    Java中的ExecutorService接口及其实现类如ThreadPoolExecutor,允许我们自定义线程池的核心参数,如核心线程数、最大线程数、线程存活时间等。线程池能有效控制并发量,防止过多线程导致系统资源耗尽,同时提供定时...

    ThreadPoolExecutor运转机制介绍

    如果当前线程池中的线程数量已经达到 `maximumPoolSize`,那么将会根据 `RejectedExecutionHandler` 的拒绝策略来处理新任务。 4. **线程的销毁**:当线程池中的线程数量大于 `corePoolSize` 时,那些超过 `...

    线程池ThreadPoolExecutor原理源码分析.md

    `ThreadPoolExecutor` 通过灵活配置核心线程数量、最大线程数量、任务队列等参数,实现了对线程的有效管理和任务的高效执行。通过对 `execute()` 方法的深入分析,我们能够更全面地理解 `ThreadPoolExecutor` 在实际...

    JDK之ThreadPoolExecutor源码分析1

    本文将深入解析ThreadPoolExecutor的execute()方法执行流程,以帮助我们理解线程池的工作原理。 当一个任务被提交到线程池,线程池的执行策略主要分为四步: 1. 首先,线程池会检查当前的核心线程数是否已达到设定...

    JAVA线程(第三版)

    《JAVA线程(第三版)》是一本深入探讨Java多线程编程的权威书籍,针对Java线程的管理和优化提供了详尽的解析。线程在现代计算机编程中扮演着至关重要的角色,尤其是在并发处理和高性能应用中。Java以其强大的线程...

    Java系列深入解析Java多线程

    - **ThreadPoolExecutor**:最常见的线程池实现,可以配置核心线程数、最大线程数、线程存活时间等参数。 - **Executors工厂方法**:提供newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool和...

    Java 多线程编程面试集锦20道问题解答Java多线程编程高难度面试题及解析

    9. **线程间通信**:线程间通信通过共享数据或使用wait/notify机制协调工作。例如,生产者消费者模型、哲学家就餐问题等。 10. **线程优先级**:Java线程有优先级,但具体效果依赖于操作系统。线程优先级高的线程在...

    java多线程面试题59题集合

    - ThreadPoolExecutor:最常见的线程池实现,可以定制线程池大小、队列容量、拒绝策略等。 - Executors:静态工厂方法,提供预定义的线程池,如FixedThreadPool、SingleThreadExecutor、CachedThreadPool等。 6. ...

    多线程下载原理

    Java提供了`Thread`类来创建线程,`ExecutorService`和`ThreadPoolExecutor`等工具来管理线程池,以及`Future`接口来获取线程执行的结果。开发者通常会定义一个下载任务类,包含文件块的URL和本地保存路径,然后创建...

    多线程下载demo

    线程池的核心组件包括工作队列、线程池大小、核心线程数、最大线程数、空闲线程存活时间等。在多线程下载中,线程池用于调度多个下载任务,确保系统资源的有效利用。例如,Java中的`ExecutorService`接口及其实现如`...

    多线程详解(通俗易懂)

    Java的`ExecutorService`和`ThreadPoolExecutor`类实现了线程池功能,允许开发者根据需求调整线程池的大小、任务队列和拒绝策略等参数,以优化性能和资源利用。 最后,我们提一下死锁,这是一种线程间相互等待对方...

    线程池源码解析-多线程

    在实际开发中,结合`ThreadPoolExecutor`的其他方法,如`execute()`、`shutdown()`、`shutdownNow()`等,可以灵活地控制线程池的工作流程,以应对不同的并发场景。总之,掌握线程池源码不仅有助于优化代码,还能在...

    多线程数据采集器源码

    1. **线程池管理**:为了有效控制和管理线程,源码可能会使用Java的`ExecutorService`或者Python的`concurrent.futures.ThreadPoolExecutor`等线程池实现。线程池可以预先创建一定数量的线程,当有任务需要执行时,...

    Java/Android关于线程使用的总结

    ### Java/Android线程使用深度解析 在计算机科学领域,线程与进程是核心概念,尤其是在多任务操作系统中。本文将深入探讨Java/Android环境下的线程管理,包括线程的创建、线程池的利用及`ThreadHandler`的运用等...

    多线程笔记_java/多线程_

    本文档将全面解析多线程的基础知识,从简单到复杂,帮助开发者深入理解并掌握这一核心技术。 一、多线程基础 1.1 线程与进程 线程是操作系统分配处理器时间的基本单元,而进程则是资源分配的基本单位。一个进程...

    Java线程-第三版(CHM电子版)

    5. **并发集合**:讲解线程安全的集合类,如`ConcurrentHashMap`、`CopyOnWriteArrayList`和`BlockingQueue`等,以及它们在多线程环境下的使用策略。 6. **死锁与活锁**:分析线程间的同步问题可能导致的死锁和活锁...

Global site tag (gtag.js) - Google Analytics