zoukankan      html  css  js  c++  java

  • 多线程（六）

    InvokeAny 以及 InvokeAll 解释清楚以后，我们最后的一部是关于executor的执行的过程，这个实在线程池中执行的，要想弄明白executor的执行的过程，必须对java对应的线程池进行说明，一般的线程池在前面已经说过，通过Executors创建 executorService接口，真正执行的是我们在（五）里面说的abstractExecutorService的子类ThreadPoolExecutor的方法：

    java.util.concurrent.ThreadPoolExecutor.execute(Runnable)

    首先必须了解线程池的概念。

    诸如web服务器、数据库服务器、文件服务器和邮件服务器等许多服务器应用都面向处理来自某些远程来源的大量短小的任务。构建服务器应用程序的一个过于简单的模型是：每当一个请求到达就创建一个新的服务对象，然后在新的服务对象中为请求服务。但当有大量请求并发访问时，服务器不断的创建和销毁对象的开销很大。所以提高服务器效率的一个手段就是尽可能减少创建和销毁对象的次数，特别是一些很耗资源的对象创建和销毁，这样就引入了“池”的概念，“池”的概念使得人们可以定制一定量的资源，然后对这些资源进行复用，而不是频繁的创建和销毁。

    线程池是预先创建线程的一种技术。线程池在还没有任务到来之前，创建一定数量的线程，放入空闲队列中。这些线程都是处于睡眠状态，即均为启动，不消耗CPU，而只是占用较小的内存空间。当请求到来之后，缓冲池给这次请求分配一个空闲线程，把请求传入此线程中运行，进行处理。当预先创建的线程都处于运行状态，即预制线程不够，线程池可以自由创建一定数量的新线程，用于处理更多的请求。当系统比较闲的时候，也可以通过移除一部分一直处于停用状态的线程。

    线程池的作用：

    线程池作用就是限制系统中执行线程的数量。根据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果；少了浪费了系统资源，多了造成系统拥挤效率不高。用线程池控制线程数量，其他线程排队等候。一个任务执行完毕，再从队列的中取最前面的任务开始执行。若队列中没有等待进程，线程池的这一资源处于等待。当一个新任务需要运行时，如果线程池中有等待的工作线程，就可以开始运行了；否则进入等待队列。

    为什么要用线程池:

    1.减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务。

    2.可以根据系统的承受能力，调整线程池中工作线线程的数目，防止因为消耗过多的内存，而把服务器累趴下(每个线程需要大约1MB内存，线程开的越多，消耗的内存也就越大，最后死机)。

    线程池的注意事项

    虽然线程池是构建多线程应用程序的强大机制，但使用它并不是没有风险的。在使用线程池时需注意线程池大小与性能的关系，注意并发风险、死锁、资源不足和线程泄漏等问题。

    （1）线程池大小。多线程应用并非线程越多越好，需要根据系统运行的软硬件环境以及应用本身的特点决定线程池的大小。一般来说，如果代码结构合理的话，线程数目与CPU 数量相适合即可。如果线程运行时可能出现阻塞现象，可相应增加池的大小；如有必要可采用自适应算法来动态调整线程池的大小，以提高CPU 的有效利用率和系统的整体性能。

    （2）并发错误。多线程应用要特别注意并发错误，要从逻辑上保证程序的正确性，注意避免死锁现象的发生。

    （3）线程泄漏。这是线程池应用中一个严重的问题，当任务执行完毕而线程没能返回池中就会发生线程泄漏现象。

    简单线程池的设计

    一个典型的线程池，应该包括如下几个部分：
    1、线程池管理器（ThreadPool），用于启动、停用，管理线程池
    2、工作线程（WorkThread），线程池中的线程
    3、请求接口（WorkRequest），创建请求对象，以供工作线程调度任务的执行
    4、请求队列（RequestQueue）,用于存放和提取请求
    5、结果队列（ResultQueue）,用于存储请求执行后返回的结果

    线程池管理器，通过添加请求的方法（putRequest）向请求队列（RequestQueue）添加请求，这些请求事先需要实现请求接口，即传递工作函数、参数、结果处理函数、以及异常处理函数。之后初始化一定数量的工作线程，这些线程通过轮询的方式不断查看请求队列（RequestQueue），只要有请求存在，则会提取出请求，进行执行。然后，线程池管理器调用方法（poll）查看结果队列（resultQueue）是否有值，如果有值，则取出，调用结果处理函数执行。通过以上讲述，不难发现，这个系统的核心资源在于请求队列和结果队列，工作线程通过轮询requestQueue获得人物，主线程通过查看结果队列，获得执行结果。因此，对这个队列的设计，要实现线程同步，以及一定阻塞和超时机制的设计，以防止因为不断轮询而导致的过多cpu开销。

    java中的线程池

    Java里面线程池的顶级接口是Executor，但是严格意义上讲Executor并不是一个线程池，而只是一个执行线程的工具。真正的线程池接口是ExecutorService。

    比较重要的几个类：

    ExecutorService                     真正的线程池接口。

    ScheduledExecutorService       能和Timer/TimerTask类似，解决那些需要任务重复执行的问题。

    ThreadPoolExecutor                 ExecutorService的默认实现。

    ScheduledThreadPoolExecutor    继承ThreadPoolExecutor的ScheduledExecutorService接口实现，周期性任务调度的类实现。

    要配置一个线程池是比较复杂的，尤其是对于线程池的原理不是很清楚的情况下，很有可能配置的线程池不是较优的，因此在Executors类里面提供了一些静态工厂，生成一些常用的线程池。

    1. newSingleThreadExecutor

    创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

    2.newFixedThreadPool

    创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

    3. newCachedThreadPool

    创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，

    那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。

    4.newScheduledThreadPool

    创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。

    从上面四个主要线程池的创建函数的源代码：

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

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

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

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
          new DelayedWorkQueue());
}

    可以知道，真正执行的是在ThreadPollExecutor，我们的主要任务是关于executor的执行的过程，首先ThreadPoolExecutor 这个类的结构：

    线程运行的状态以及中间使用的变量：

    volatile int runState; 
static final int RUNNING = 0; 
static final int SHUTDOWN = 1; 
static final int STOP = 2; 
static final int TERMINATED = 3;
    //核心线程池大小 
private volatile int corePoolSize; 
//线程池允许的最大线程数量 
private volatile int maximumPoolSize; 
//当前线程池中的线程数量 
private volatile int poolSize;

    上面我们说过了简单线程池的设计，对应的部分都能在ThreadPoolExecutor找到。

    工作线程：

    /**
     * The queue used for holding tasks and handing off to worker
     * threads.  Note that when using this queue, we do not require
     * that workQueue.poll() returning null necessarily means that
     * workQueue.isEmpty(), so must sometimes check both. This
     * accommodates special-purpose queues such as DelayQueues for
     * which poll() is allowed to return null even if it may later
     * return non-null when delays expire.
     */
    private final BlockingQueue<Runnable> workQueue;

    现在可以揭开异步执行方法executor(Runnable command)的真面目了，回到线程池执行器ThreadPoolExecutor，乍看这个方法，蛮精干的：

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
}

     

    在分析之前，需要再解释几个概念，ThreadPoolExecutor是一个线程池执行器，它里面维护了一个线程池，以及一个等待执行的任务队列。

          所谓线程池，其实就是一个Worker对象的集合：HashSet<Worker> workers = new HashSet<Worker>();至于这个Worker的基本类图如下：

         从类图看(当然最好结合代码)，Worker本身是一个Runnable，它自己维护了执行它的线程对象thead，又维护了一个Runnable对象firstTask（这个对象就是ThreadPoolExecutor线程池的任务对象了），当ThreadPoolExecutor执行一个任务的时候，先获得 (最直接的方法就是new)一个可用的Thread对象，然后再获得(最直接方法就是new)一个Worker对象，并把Thread对象包装进这个Worker对象中，接着让这个thread对象start就开始执行这个Worker对象的run()方法，而run()方法中会去执行Worker.firstTask.run()方法。这就间接的的执行了目标任务，同时通过worker这个包装(或者说代理)之后，可用做很多额外的工作，比如中断自身执行线程，记录在该线程上执行过的任务数量等。分析addIfUnderCorePoolSize(command)方法时还会分析Worker类。

        线程池数量poolSize指工作线程Worker对象的集合workers的实际大小，通过workers.size()可直接获得。            

          核心线程池数量corePoolSize，可理解为工作线程Worker对象的集合workers的目标大小。如果poolSize> corePoolSize，那么ThreadPoolExecutor就会有机制在适当的时候回收闲置的线程。

         最大线程池数量maxPoolSize，就是工作线程Worker对象的集合workers的大小上限。假如说任务队列满了，再来新任务时，若poolSize还没达到maxPoolSize，则继续创建新的线程来执行新任务，若不幸poolSize达到了上限maxPoolSize，那不能再创建新的线程了，只能采取reject策略来拒绝新任务。

    所谓任务队列，就是一个Runnable对象的阻塞队列：BlockingQueue<Runnable> workQueue; 可根据不同需求设置不同的队列类型。

    下面分析execute(command)执行流程：

    1. 第四行if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))

    解释： 如果当前线程池中线程数量poolSize >=  核心线程数量corePoolSize 成立，那么逻辑或运算符后面的方法addIfUnderCorePoolSize(command)就忽略不做，而直接进花括号内部；如果poolSize >= corePoolSize不成立，尝试调用addIfUnderCorePoolSize(command)方法，该方法返回true就进花括号，否则整个execute方法就结束。

    理解：这很好理解，如果当前线程数量poolSize>=核心线程数量corePoolSize，那当然无法再把当前任务加入到核心线程池中执行了，于是进花括号选择其他的策略执行；如果poolSize没有达到corePoolSize，那很自然是把当前任务放到核心线程池执行，也就是执行逻辑或运算符后的方法addIfUnderCorePoolSize(command)。“放到核心线程池执行”是什么意思呢？就是new 一个新工作线程放到workers集合中，让这个新线程来执行当前的任务command，而这个新线程可以认为是核心线程池中的其中一个线程。

    addIfUnderCorePoolSize(command)方法做了什么事情？请看代码：

    private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }

    主要就是在poolSize < corePoolSize并且当前状态runState == RUNNING时通过方法addThread(firstTask)返回一个线程t，然后马上t.start()执行任务，返回true。这里有失败的可能，因为虽然在前面execute方法中已经保证了poolSize < corePoolSize进入该方法，但是当时并没有加锁，很有可能到了当前的位置poolSize已经改变了，所以这里必须再次检查并且必须加锁访问。顺便简单看下addThread方法：

    private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);
            int nt = ++poolSize;
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

    直接new一个工作线程Worker对象，封装目标任务firstTask，用线程工厂创建一个新线程t,然后t又赋值给Worker的thread属性，这下对Worker有了新理解吧？线程t 用来执行Worker对象，而Worker对象的thread属性的值＝t ，他维护了执行它自己的那个线程。t.start()便开始执行worker.run()，而worker.run中会去执行封装目标任务firstTask.run()方法。当然Worker还没有完，后面还会分析到。

    addIfUnderCorePoolSize(command)会返回false吗？如果corePoolSize设置的足够大，基本就不会失败，那execute(command)方法做完这句if语句判断就结束了，压根不需要进入花括号继续。但是不幸的是，addIfUnderCorePoolSize还是经常会失败的。所以接下来还得看看if大括号里面的逻辑。

    2. 第五行if (runState == RUNNING && workQueue.offer(command))

    解释：runState表示这个TheadPoolExecutor的状态，可以有4个状态

    a) RUNNING可接收新任务并执行任务队列

            b) SHUTDOWN不能接收新任务，但可以继续执行任务队列

            c) STOP不能接收新任务，也不在处理任务队列，并且中断正在执行的任务

    d)      TERMINATED在STOP基础上，所有线程都已终止

    程序若到了这一步，说明当前线程数量poolSize >=核心线程数量corePoolSize，这里先判断是不是这个TheadPoolExecutor还是RUNNING状态，若是则试着加入到任务队列workQueue中，无法加入的唯一可能就是队列已经满了。先说没满的情况，就是任务加入到任务队列成功。按照常理，加入了队列以后，只要保证有工作线程就ok了，工作线程会自动去执行任务队列的。所以判断一下if ( runState != RUNNING || poolSize == 0),在这个if为true时候，去保证一下任务队列有线程会执行，即执行ensureQueuedTaskHandled(command)方法。这里有两种情况，情况一：runState != RUNNING，这种情况在ensureQueuedTaskHandled方法中会把任务丢给reject拒绝策略处理，情况二：poolSize == 0，这种情况是new一个新线程加入到工作线程集合workers中。

    3. 第九行else if (!addIfUnderMaximumPoolSize(command))

    解释：程序执行到这个分支，说明上面第五行if条件为false，也就是说当前状态runState ！= RUNNING，或者任务队列workQueue已经满了。先看第一个条件下，前面解释过runState，除了RUNNING状态，其他三个状态都不能接收新任务，所以当runState ！= RUNNING时新任务只能根据reject策略拒绝，而这个拒绝的逻辑是在addIfUnderMaximumPoolSize方法中实现的；再看第二个条件下，workQueue已经满，潜在的条件是runState == RUNNING，这种情况怎么处理新任务呢？很简单，若当前线程数量已经poolSize没有达到最大线程数量maxPoolSize，则创建新的线程去执行这个无法加入任务队列的新任务，否则就根据reject策略拒绝，这里的拒绝逻辑就在这个else if条件成立的子句中做的，即第10行reject(command);

    到此，任务异步执行的整个过程execute(Runnable command)分析完毕，顺便画个流程图作为本文的结尾，本来应该呼应一下前文，完成对工作线程Worker类的分析，包括选取执行任务队列、及在任务队列为空时将Worker对象回收，但是看来得放到下一篇了。

    本文参考：

    http://www.cnblogs.com/coser/archive/2012/03/10/2389264.html

    http://www.oschina.net/question/565065_86540
  • 相关阅读:
    Ubuntu环境下IPython的搭建和使用
    智能移动导游解决方案简介
    企业文化、团队文化与知识共享
    CoinPunk项目介绍
    Insight API开源项目介绍
    比特币Bitcoin源代码安装编译
    Javascript单元测试Unit Testing之QUnit
    Node.js的UnitTest单元测试
    Node.js调试
    Alfresco 4 项目介绍
  • 原文地址：https://www.cnblogs.com/zhailzh/p/3986427.html
Copyright © 2011-2022 走看看