线程池代码完整剖析（二）

线程池代码完整剖析（二）

一、概述

 

上一篇讲了如果启动一个组件的抽象类，下面具体的说明Jetty线程池是如何实现的，本文是根据源代码还有具体的注释讲述。

 

二、ThreadPool接口

 

请看源代码：

 

 

public interface ThreadPool { 
	/* 从线程池中取出一个线程分发任务 */
	public abstract boolean dispatch(Runnable job);

	/** * Blocks until the thread pool is {@link LifeCycle#stop stopped}. */
	public void join() throws InterruptedException; 

	/** * 返回当前线程池线程的总数 */
	public int getThreads();

	/** * 返回线程池中空闲线程数目 */
	public int getIdleThreads();

	/** * 是否当前的线程池中的线程很少 */
	public boolean isLowOnThreads();
} 

 

该接口的每一个方法在注释上都说明了，这里不再详细说明

 

三、QueuedThreadPool

 

QueuedThreadPool是ThreadPool接口的实现类，该类也继承了AbstractLifeCycle. 该类需要注意的就是线程池维护一个基于数组的循环队列，来保存请求任务。

 

 

下面说明该类定义的成员变量：

这里有个定义一定要搞清数组和队列是同一个词义，因为该类的队列是由数组组成

    private String _name;//有界的线程池名字
   private Set _threads;//线程池中线程集合
    /*
     * 线程池中闲散线程集合
     **/
    private List _idle;

   private Runnable[] _jobs;//有界队列是由循环数组组成，也是用来保存请求任务（当线程池中的线程达到了一定数量时）
    private int _nextJob;//数组指针，指向当前执行任务，数组当中是一个循环数组
    private int _nextJobSlot;//有界队列累计的任务数量指针

 

_nextJob,_nextJobSlot变量十分有用必须要理解含义是什么

    private int _queued;//有界队列当前任务的数目，还有这么多数量的任务没有处理
    private int _maxQueued;//队列中容纳最大的任务数量

 

 /**
     * 此锁是同步闲散线程集合以及一些公共变量
     */
    private final Object _lock = new Lock();
    /****
     * 这个锁是为了保持线程池中保存线程集合操作的同步问题
     */
    private final Object _threadsLock = new Lock();//这个是线程池的锁
    private final Object _joinLock = new Lock();

 

上面的三个锁对象地位非常的重要，主要目的是在多线程中涉及到变量共享的问题，如何进行有效的同步。这样的设计缩小了同步代码块也就是说减少了不必要锁的竞争，提高了系统的效率，读者在读源码的时候要清醒的理解每个同步代码块下涉及的变量，设计者是如何同步的，这样的设计在工作当中会经常的运用。。。

 

    private long _lastShrink;//根据这个变量缩减线程池的数目
    private int _maxIdleTimeMs=60000;//线程最大的闲散时间
    private int _maxThreads=250;//线程池中最大的线程数目
    private int _minThreads=2;//线程池中最小的线程数目
    private boolean _warned=false;
    private int _lowThreads=0;//线程池线程数目最低阀值
    private int _priority= Thread.NORM_PRIORITY;//得到线程池中线程的优先权
    //线程池中线程数是扩大还是缩小，以及是否要新建一个线程处理该任务还是把请求的任务放在队列中
    private int _spawnOrShrinkAt=0;
    private int _maxStopTimeMs;//线程最大的等待时间，如何超出该时间段就死亡

 

该类的成员变量描述完后，就已经知道他们的用图，现在要分三条线路给大家展示线程池的秘密；

 

第一天线程池的是如何初始化

 

第二条线路是用户请求的任务，而线程池是如何分配的线程给请求用户或者说是以什么样的策略方式

 

第三条线程池中的线程的生命周期

 

我认为理解三条思路就理解了设计者的精髓，下面分别阐明这三条思路

 

四、线程池如何初始化

 

首先要说start方法，然后再说doStart()方法，代码如下所示：

 

 

    public final void start() throws Exception
    {
        synchronized (this)
        {
            try
            {
                if (_state == STARTED || _state == STARTING)
                    return;
                setStarting();
                doStart();
                Log.debug("started {}",this);
                setStarted();
            }
            catch (Exception e)
            {
                setFailed(e);
                throw e;
            }
            catch (Error e)
            {
                setFailed(e);
                throw e;
            }
        }
    }

 

该方法主要是启动线程池组件如：setStarting()和setStarted()方法，而调用doStart()方法，是初始化线程池的主要函数,代码如下：

 /* * 新建线程池的集合，新建线程池中闲散的线程，新建有界队列，
     * 为线程池创建最小的线程数目
     * Construct the minimum number of threads.
     */
    protected void doStart() throws Exception
    {
        if (_maxThreads<_minThreads || _minThreads<=0)
            throw new IllegalArgumentException("!0<minThreads<maxThreads");
        
        _threads=new HashSet();
        _idle=new ArrayList();
        _jobs=new Runnable[_maxThreads];
        
        for (int i=0;i<_minThreads;i++)
        {
            newThread();
        }   
    }

 

建立了保存线程池中线程的集合，以及保存线程池闲散线程的集合和保存任务请求的数组队列，同时创建了线程池限定最小数目的线newThread方法如下所示：

 

    protected void newThread()
    {
        synchronized (_threadsLock)
        {
            if (_threads.size()<_maxThreads)
            {
                PoolThread thread =new PoolThread();
                _threads.add(thread);
                thread.setName(thread.hashCode()+"@"+_name+"-"+_id++);
                thread.start(); 
            }
            else if (!_warned)    
            {
                _warned=true;
                Log.debug("Max threads for {}",this);
            }
        }
    }

 

PoolThread是QueuePoolThread类的内部类，请看下面进行分析，这里就不分析主要作用。把创建的线程放在集合里面，然后启动该线程，下面稍微说明一下启动该线程的run方法，这里只说明该方法的部分代码：

 

 public void run()
        {
            boolean idle=false;
            Runnable job=null;
            try
            {
                while (isRunning())
                {   
                    // Run any job that we have.
                    if (job!=null)
                    {
                        final Runnable todo=job;
                        job=null;
                        idle=false;//这个标志等任务调度完后，在放到闲散线程集合里
                        todo.run();
                    }
                    
                    synchronized(_lock)
                    {
                        // is there a queued job?
                        if (_queued>0)
                        {
                            _queued--;
                            job=_jobs[_nextJob++];
                            if (_nextJob==_jobs.length)
                                _nextJob=0;
                            continue;
                        }

                        // Should we shrink?
                        final int threads=_threads.size();
                        if (threads>_minThreads && 
                            (threads>_maxThreads || 
                             _idle.size()>_spawnOrShrinkAt))   
                        {
                        	
                            long now = System.currentTimeMillis();
                            if ((now-_lastShrink)>getMaxIdleTimeMs())
                            {
                                _lastShrink=now;
                                _idle.remove(this);
                                return;
                            }
                        }

                        /**
                         * 线程池中在刚刚在池中创建线程的时候没有任务安排
                         * 所以同时也是闲散线程的数目
                         */
                        if (!idle)
                        {   
                            // Add ourselves to the idle set.
                            _idle.add(this);
                            
//                            System.out.println(_idle.size());
                            idle=true;
                        }
                    }

                    // We are idle
                    // wait for a dispatched job
                    synchronized (this)
                    {
                        if (_job==null)
                            this.wait(getMaxIdleTimeMs());
                        job=_job;
                        _job=null;
                    }
                }
            }

            catch (InterruptedException e)
            {
                Log.ignore(e);
            }
            finally
            {
                synchronized (_lock)
                {
                    _idle.remove(this);
                }
                synchronized (_threadsLock)
                {
                    _threads.remove(this);
                }
                synchronized (this)
                {
                    job=_job;
                }
                
                // we died with a job! reschedule it
                if (job!=null)
                {
                    QueuedThreadPool.this.dispatch(job);
                }
            }
        }
        

 

上面有一个while循环，始终保持线程是运行状态，这里只想说的代码是标记红色的，因为刚刚创建线程时候只会运行红色的代码块（刚刚开始线程是闲散的所以添加闲散集合中），而且执行到this.wait(getMaxIdleTimeMs());这行代码，该线程就会阻塞，目的是等待有任务的线程去唤醒他。下面会详细说明的。

 

第二三条线路请看续集。。。