threadPoolExecutor代码

iamzhongyong

浏览: 796373 次
性别:
来自: 杭州

最近访客更多访客>>

wtmiao000

顾惜朝

kelly_zsl

u012363178

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

java

java.util.concurrent包的出现，为我们实现多线程并发编程带来了很大的方便，但是同时也有很多问题，如果对于原理缺乏了解，使用起来可能就会有一些点忽略掉，导致应用出现问题，关于线程池已经想看源代码很长时间了，不能再拖下去了，本来打算是从头自己看源代码，但是发现一个哥们的博客，上面讲的挺详细，于是打算借鉴一些过来，然后加入自己的理解，不再重复造轮子了。

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

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

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

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

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

（5）newSingleThreadScheduledExecutor：创建一个单线程的线程池。此线程池支持定时以及周期性执行任务的需求。

首先看ThreadPoolExecutor的构造函数中的核心，对于各个参数有必要介绍一下。

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;
    }

（1）corePoolSize：core zise 大家都懂得，不解释

（2）maximumPoolsize：线程池的最大数量

（3）keepAliveTime：空闲的线程等待工作的最长时间，纳秒，和unit配合使用

（4）woreQueue：线程池处理的任务队列

（5）threadFactory：线程创建的工厂类，threadPoolExecutor中的有默认实现，开始是最好自己实现一个，并且命名，方便问题排查

（6）handler：线程池的拒绝策略，默认有四种可选择，也可自己写一个

从两个if判断里面能够看出参数缺一不可，keepAliveTime的时间通过TimeUnit转为纳秒。

ThreadFactory是一个接口，里面就一个方法，Thread newThread(Runnable r) 用来创造线程

在Executors中有一个默认的实现：DefaultThreadFactory

static class DefaultThreadFactory implements ThreadFactory {
        static final AtomicInteger poolNumber = new AtomicInteger(1);
        final ThreadGroup group;
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String namePrefix;

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null)? s.getThreadGroup() :
                                 Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            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;
        }
    }

通过默认的factory创造的线程，是非守护，并且优先级为NORM_PRIORITY，线程的名字为“pool-线程池数目-thread-线程的编号”。

对于线程池中的所有线程默认都转换为非后台线程，这样主线程退出时不会直接退出JVM，而是等待线程池结束。还有一点就是默认将线程池中的所有线程都调为同一个级别，这样在操作系统角度来看所有系统都是公平的，不会导致竞争堆积。

线程的状态

为了节省资源和有效释放资源关闭一个线程池就显得很重要。线程池Executor是异步的执行任务，因此任何时刻不能够直接获取提交的任务的状态。这些任务有可能已经完成，也有可能正在执行或者还在排队等待执行。因此关闭线程池可能出现一下几种情况：

（1）平缓关闭：已经启动的任务全部执行完毕，同时不再接受新的任务

（2）立即关闭：取消所有正在执行和未执行的任务

线程池的状态，在源代码里面有以下注释，感觉比任务文章来的给力点呵呵

RUNNING	接收新的任务，同时处理队列中已有的任务
SHUTDOWN	不再接受新的任务，但是处理队列中的任务
STOP	不接受新任务，不处理队列中任务，同时终止正在运行的任务
TERMINATED	和stop一样，但是所有的线程终止

RUNNING到SHUTDOWN：调用shutdown()方法

RUNNING or SHUTDOWN 到STOP ：调用shutdownNow()方法

SHUTDOWN 到 TERMINATED：任务队列和线程池都空时

STOP 到 TERMINATED ：线程池为空时

线程池的拒绝策略

这四种策略是独立无关的，是对任务拒绝处理的四中表现形式。最简单的方式就是直接丢弃任务。但是却有两种方式，到底是该丢弃哪一个任务，比如可以丢弃当前将要加入队列的任务本身（DiscardPolicy）或者丢弃任务队列中最旧任务（DiscardOldestPolicy）。丢弃最旧任务也不是简单的丢弃最旧的任务，而是有一些额外的处理。除了丢弃任务还可以直接抛出一个异常（RejectedExecutionException），这是比较简单的方式。抛出异常的方式（AbortPolicy）尽管实现方式比较简单，但是由于抛出一个RuntimeException，因此会中断调用者的处理过程。除了抛出异常以外还可以不进入线程池执行，在这种方式（CallerRunsPolicy）中任务将有调用者线程去执行。

四种策略均实现了RejectedExecutionHandler接口

方法rejectedExecution()在线程池不能接收任务的时候会被ThreadPoolExecutor调用，参数runnable是需要被执行的任务，executor是线程池对象。

public interface RejectedExecutionHandler {

void rejectedExecution(Runnable r, ThreadPoolExecutor executor);

}

除了上述的几种拒绝策略外，还可以自己实现RejectedExecutionHandler接口，例如如下拒绝策略，当任务没法加入任务对队列时，阻塞，直到可以放入队列中。

ejectedExecutionHandler handler = new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable task,ThreadPoolExecutor executor) {
           // 如果处理不过, 打日志, 并阻塞,
           logger.error("The thread pool is full and be blocked!"
                 + " corePoolSize=" + executor.getCorePoolSize()
                 + ", maxPoolSize=" + executor.getMaximumPoolSize()
                 + ", keepAliveTime="
                 + executor.getKeepAliveTime(TimeUnit.SECONDS)
                 + ", poolSize=" + executor.getPoolSize()
                 + ", activeCount=" + executor.getActiveCount()
                 + ", taskCount=" + executor.getTaskCount()
                 + ", largestPoolSize=" + executor.getLargestPoolSize()
                 + ", completedTaskCount="
                 + executor.getCompletedTaskCount());
 
           try {
              executor.getQueue().put(task);
           } catch (InterruptedException e) {
              logger.error("put the task into queue is failure!", e);
           }
        }
     };

关于Worker对象

对于ThreadPoolExecutor而言，一个线程就是一个Worker对象，它与一个线程绑定，当Worker执行完毕就是线程执行完毕。

 /**
         * Runs a single task between before/after methods.
         */
        private void runTask(Runnable task) {
            final ReentrantLock runLock = this.runLock;
            runLock.lock();
            try {
                /*
                 * Ensure that unless pool is stopping, this thread
                 * does not have its interrupt set. This requires a
                 * double-check of state in case the interrupt was
                 * cleared concurrently with a shutdownNow -- if so,
                 * the interrupt is re-enabled.
                 */
                if (runState < STOP &&
                    Thread.interrupted() &&
                    runState >= STOP)
                    thread.interrupt();
                /*
                 * Track execution state to ensure that afterExecute
                 * is called only if task completed or threw
                 * exception. Otherwise, the caught runtime exception
                 * will have been thrown by afterExecute itself, in
                 * which case we don't want to call it again.
                 */
                boolean ran = false;
                beforeExecute(thread, task);
                try {
                    task.run();
                    ran = true;
                    afterExecute(task, null);
                    ++completedTasks;
                } catch (RuntimeException ex) {
                    if (!ran)
                        afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                runLock.unlock();
            }
        }

        /**
         * Main run loop
         */
        public void run() {
            try {
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);
            }
        }

执行一个给定的任务，任务可能会在一个新的线程中执行，也有可能在线程池中已近存在的线程中执行。如果任务不能被提交执行（可能是因为线程池shutdown或者任务队列已经满了），此时这个任务会被handle来处理

几种常见的死锁

（1）锁顺序死锁(lock-ordering deadlock)：多个线程试图通过不同的顺序获得多个相同的资源，则发生的循环锁依赖现象。

（2）动态的锁顺序死锁（Dynamic Lock Order Deadlocks）：多个线程通过传递不同的锁造成的锁顺序死锁问题。

（3）资源死锁（Resource Deadlocks）：线程间相互等待对方持有的锁，并且谁都不会释放自己持有的锁发生的死锁。也就是说当现场持有和等待的目标成为资源，就有可能发生此死锁。这和锁顺序死锁不一样的地方是，竞争的资源之间并没有严格先后顺序，仅仅是相互依赖而已。

锁顺序死锁

最经典的锁顺序死锁就是LeftRightDeadLock.

public class LeftRightDeadLock {
    final Object left = new Object();
    final Object right = new Object();
    public void doLeftRight() {
        synchronized (left) {
            synchronized (right) {
                execute1();
            }
        }
    }
    public void doRightLeft() {
        synchronized (right) {
            synchronized (left) {
                execute2();
            }
        }
    }
    private void execute2() {
    }
    private void execute1() {
    }
}

这个例子很简单，当两个线程分别获取到left和right锁时，互相等待对方释放其对应的锁，很显然双方都陷入了绝境。

动态的锁顺序死锁

与锁顺序死锁不同的是动态的锁顺序死锁只是将静态的锁变成了动态锁。一个比较生动的例子是这样的。

public void transferMoney(Account fromAccount,//
        Account toAccount,//
        int amount
        ) {
    synchronized (fromAccount) {
        synchronized (toAccount) {
            fromAccount.decr(amount);
            toAccount.add(amount);
        }
    }
}

当我们银行转账的时候，我们期望锁住双方的账户，这样保证是原子操作。看起来很合理，可是如果双方同时在进行转账操作，那么就有可能发生死锁的可能性。

很显然，动态的锁顺序死锁的解决方案应该看起来和锁顺序死锁解决方案差不多。但是一个比较特殊的解决方式是纠正这种顺序。例如可以调整成这样：

Object lock = new Object();
public void transferMoney(Account fromAccount,//
        Account toAccount,//
        int amount
        ) {
    int order = fromAccount.name().compareTo(toAccount.name());
    Object lockFirst = order>0?toAccount:fromAccount;
    Object lockSecond = order>0?fromAccount:toAccount;
    if(order==0){
        synchronized(lock){
            synchronized(lockFirst){
                synchronized(lockSecond){
                    //do work
                }
            }
        }
 
    }else{
        synchronized(lockFirst){
            synchronized(lockSecond){
                //do work
            }
        }
    }
}

这个挺有意思的。比较两个账户的顺序，保证此两个账户之间的传递顺序总是按照某一种锁的顺序进行的，即使多个线程同时发生，也会遵循一次操作完释放完锁才进行下一次操作的顺序，从而可以避免死锁的发生。

资源死锁

资源死锁比较容易理解，就是需要的资源远远大于已有的资源，这样就有可能线程间的资源竞争从而发生死锁。一个简单的场景是，应用同时从两个连接池中获取资源，两个线程都在等待对方释放连接池的资源以便能够同时获取到所需要的资源，从而发生死锁。

资源死锁除了这种资源之间的直接依赖死锁外，还有一种叫线程饥饿死锁（thread-starvation deadlock）。严格意义上讲，这种死锁更像是活跃度问题。例如提交到线程池中的任务由于总是不能够抢到线程从而一直不被执行，造成任务的“假死”状况。

除了上述几种问题外，还有协作对象间的死锁以及开发调用的问题。这个描述起来会比较困难，也不容易看出死锁来。

避免和解决死锁

通常发生死锁后程序难以自恢复。但也不是不能避免的。有一些技巧和原则是可以降低死锁可能性的。

最简单的原则是尽可能的减少锁的范围。锁的范围越小，那么竞争的可能性也越小。尽快释放锁也有助于避开锁顺序。如果一个线程每次最多只能够获取一个锁，那么就不会产生锁顺序死锁。

如何设置最有的线程数量

并行的任务增加资源显然能够提高性能，但是如果是串行的任务，增加资源并不一定能够得到合理的性能提升。 Amdahl定律描述的在一个系统中，增加处理器资源对系统行的提升比率。假定在一个系统中，F是必须串行化执行的比重，N是处理器资源，那么随着N的增加最多增加的加速比：

理论上，当N趋近于无穷大时，加速比最大值无限趋近于1/F。这意味着如果一个程序的串行化比重为50%，那么并行化后最大加速比为2倍。

加速比除了可以用于加速的比率外，也可以用于衡量CPU资源的利用率。如果每一个CPU的资源利用率为100%，那么CPU的资源每次翻倍时，加速比也应该翻倍。事实上，在拥有10个处理器的系统中，程序如果有10%是串行化的，那么最多可以加速1/(0.1+(1-0.1)/10)=5.3倍，换句话说CPU的利用率只用5.3/10=53%。而如果处理器增加到100倍，那么加速比为9.2倍，也就是说CPU的利用率只有个9.3%。

假设引入的多线程都用于计算，那么性能一定会有很大的提升么？其实引入多线程以后也会引入更多的开销。

切换上下文

如果可运行的线程数大于CPU的内核数，那么OS会根据一定的调度算法，强行切换正在运行的线程，从而使其它线程能够使用CPU周期。

切换线程会导致上下文切换。线程的调度会导致CPU需要在操作系统和进程间花费更多的时间片段，这样真正执行应用程序的时间就减少了。另外上下文切换也会导致缓存的频繁进出，对于一个刚被切换的线程来说，可能由于高速缓冲中没有数据而变得更慢，从而导致更多的IO开销。

内存同步

不同线程间要进行数据同步，synchronized以及volatile提供的可见性都会导致缓存失效。线程栈之间的数据要和主存进行同步，这些同步有一些小小的开销。如果线程间同时要进行数据同步，那么这些同步的线程可能都会受阻。

阻塞

当发生锁竞争时，失败的线程会导致阻塞。通常阻塞的线程可能在JVM内部进行自旋等待，或者被操作系统挂起。自旋等待可能会导致更多的CPU切片浪费，而操作系统挂起则会导致更多的上下文切换。

了解了性能的提升的几个方面，也了解性能的开销后，应用程序就要根据实际的场景进行取舍和评估。没有一劳永逸的优化方案，不断的进行小范围改进和调整是提高性能的有效手段。当前一些大的架构调整也会导致较大的性能的提升。

简单的原则是在保证逻辑正确的情况小，找到性能瓶颈，小步改进和优化。

对于多线程编程，源码的分析一下文章总结的很齐全

分享到：

【转】主流分词器对比 | 类加载分析方案以及预防办法

2012-04-15 19:43
浏览 10089
评论(0)
分类:互联网
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论