池化技术之Java线程池

jcj_2012

浏览: 65838 次
性别:
来自: 上海

最近访客更多访客>>

cyc315

jieke456

lost-java

lvye351

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

池化技术

作用

线程池，通过复用线程来提升性能;

背景

线程是一个操作系统概念。操作系统负责这个线程的创建、挂起、运行、阻塞和终结操作。而操作系统创建线程、切换线程状态、终结线程都要进行CPU调度，这是一个耗费时间和系统资源的事情。

场景描述

例如处理某一次请求的时间是非常短暂的，但是请求数量是巨大的。如果为每个请求都单独创建一个线程，

（1）那么物理机的所有资源基本上都被操作系统创建线程、切换线程状态、销毁线程这些操作所占用，用于业务请求处理的资源反而减少了。
（2）此外一些操作系统是有最大线程数量限制的。当运行的线程数量逼近操作系统是有最大线程数的时候，操作系统会变得不稳定。

结论是：我们需要限制线程数量

注：如何创建更多的线程：

每个线程都需要一个内存栈，用于存储局部变量，操作栈等信息，通过-Xss参数来调整每个线程栈大小(64位系统默认1024kb，可根据实际需要调小，比如256KB)，通过调整该参数可以创建更多的线程，不过JVM不能无限制地创建线程

最理想的处理方式

将处理请求的线程数量控制在一个范围，既保证后续的请求不会等待太长时间，又保证物理机将足够的资源用于请求处理本身。

线程池的使用

1.线程池的特点

线程池会限制创建的线程数，从而保护系统；

线程池配合队列工作，限制并发处理的任务数量，当任务超限时，通过一定的策略来处理，可避免系统因为大流量而导致崩溃-只是部分拒绝服务，还是有一部分是正常服务的。

2.线程池分类

核心线程池和最大数量线程池，线程池中线程空闲一段时间会被回收，核心线程是不会被回收的。

3.合适的线程数

(1) 建议根据实际业务情况来压测决定
(2) 利特尔法则：在一个稳定系统内，长时间观察到的平均用户数量L = 长时间观察到的有效达到率 * 平均每个用户在系统花费的时间。

针对方法2的实际情况更复杂，如：在处理超时，网络抖动会导致线程花费时间不一样。
鉴于在处理超时，网络抖动会导致线程花费时间不一样，可能造成的线程数不合理，需要考虑：超时机制，线程隔离机制，快速失败机制等来保护系统

4.Java线程池使用

Java语言为我们提供了两种基础线程池的选择：ScheduledThreadPoolExecutor和ThreadPoolExecutor。它们都实现了ExecutorService接口
注： ExecutorService接口本身和“线程池”并没有直接关系，它的定义更接近“执行器”，而“使用线程管理的方式进行实现”只是其中的一种实现方式

Java提供了ExecutorService三种实现
(1)ThreadPoolExecutor:标准线程池
(2)ScheduledThreadPoolExecutor:支持延迟任务的线程池
(3)ForkJoinPool:
类似于ThreadPoolExecutor，但是使用work-stealing模式，其会为线程池中的每个线程创建一个队列，从而用work-stealing(任务窃取)算法使得线程可以从其他线程队列里窃取任务来执行。即如果自己的任务处理完成了，可以去忙碌的工作线程哪里窃取任务执行。

5.ThreadPoolExecutor详解

介绍

ThreadPoolExecutor是JDK并发包提供的一个线程池服务，基于ThreadPoolExecutor可以很容易将一个Runnable接口的任务放入线程池中。

原理：线程池是怎样处理某一个运行任务的

首先，通过线程池提供的submit()方法或者execute()方法，要求线程池执行某个任务。线程池收到这个要求执行的任务后，会有几种处理情况：
(1) 如果当前线程池中运行的线程数量 < corePoolSize大小时，线程池会创建一个新的线程运行你的任务，无论之前已经创建的线程是否处于空闲状态。
(2) 如果当前线程池中运行的线程数量 = corePoolSize大小时，线程池会把你的这个任务加入到等待队列中。直到某一个的线程空闲了，线程池会根据设置的等待队列规则，从队列中取出一个新的任务执行。

规则如下：
根据队列规则，这个任务无法加入等待队列。这时线程池就会创建一个"非核心线程"直接运行这个任务，如果这种情况下任务执行成功，那么当前线程池中的线程数量一定大于corePoolSize。
如果这个任务，无法被“核心线程”直接执行，又无法加入等待队列，又无法创建“非核心线程”直接执行，且你没有为线程池设置RejectedExecutionHandler，这时线程池会抛出RejectedExecutionException异常，即线程池拒绝接受这个任务。

（实际上抛出RejectedExecutionException异常的操作，是ThreadPoolExecutor线程池中一个默认的RejectedExecutionHandler实现：AbortPolicy，这在后文会提到）

其次，一旦线程池中某个线程完成了任务的执行，它就会试图到任务等待队列中拿去下一个等待任务（所有的等待任务都实现了BlockingQueue接口，按照接口字面上的理解，这是一个可阻塞的队列接口），它会调用等待队列的poll()方法，并停留在哪里。

然后，当线程池中的线程超过你设置的corePoolSize参数，说明当前线程池中有所谓的“非核心线程”。那么当某个线程处理完任务后，如果等待keepAliveTime时间后仍然没有新的任务分配给它，那么这个线程将会被回收。线程池回收线程时，对所谓的“核心线程”和“非核心线程”是一视同仁的，直到线程池中线程的数量等于你设置的corePoolSize参数时，回收过程才会停止。

原ThreadPoolExecutor方法详解

查看JDK帮助文档，可以发现ThreadPoolExecutor比较简单，继承自AbstractExecutorService，而AbstractExecutorService实现了ExecutorService接口。
ThreadPoolExecutor的完整构造方法的签名是：见ThreadPoolExecutor的构建参数

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize, 
                          long keepAliveTime, 
                          TimeUnit unit, 
                          BlockingQueue<Runnable> workQueue, 
                          RejectedExecutionHandler handler) { 
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, 
         Executors.defaultThreadFactory(), handler); 
}

corePoolSize：核心线程数，会一直存活，即使没有任务，线程池也会维护线程的最少数量
maximumPoolSize：线程池维护线程的最大数量
keepAliveTime：线程池维护线程所允许的空闲时间，当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。如果 allowCoreThreadTimeout 设置为true，则所有线程均会退出直到线程数量为0。
unit：线程池维护线程所允许的空闲时间的单位、可选参数值为：TimeUnit中的几个静态属性：NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。
workQueue：线程池所使用的缓冲队列，常用的是：java.util.concurrent.ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue
handler：线程池中的数量大于maximumPoolSize，对拒绝任务的处理策略，默认值ThreadPoolExecutor.AbortPolicy()。

在JDK帮助文档中，有如此一段话：强烈建议程序员使用较为方便的Executors 工厂方法，它们均为大多数使用场景预定义了设置，如下：

（1）无界线程池，可以进行自动线程回收： Executors.newCachedThreadPool()

（2）固定大小线程池：Executors.newFixedThreadPool(int)
（3）单个后台线程：Executors.newSingleThreadExecutor()

固定大小线程池

ExecutorService newFixedThreadPool(int nThreads)

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

特点：corePoolSize和maximumPoolSize的大小是一样的,不使用keepalived,队列选择的是 LinkedBlockingQueue，该queue有一个特点，他是无界的

单线程

ExecutorService newSingleThreadExecutor()

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

特点：corePoolSize和maximumPoolSize直接设置为1

无界线程池，可以进行自动线程回收

ExecutorService newCachedThreadPool()

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

特点：
maximumPoolSize为big big。其次BlockingQueue的选择上使用SynchronousQueue。可能对于该BlockingQueue有些陌生，简单说：该QUEUE中，每个插入操作必须等待另一个线程的对应移除操作。

比如，我先添加一个元素，接下来如果继续想尝试添加则会阻塞，直到另一个线程取走一个元素，反之亦然。（想到什么？就是缓冲区为1的生产者消费者模式^_^）

排队的三种通用策略

1.直接提交。
工作队列的默认选项是SynchronousQueue，
它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。
此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

2.无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

3.有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

....

总结：

    ThreadPoolExecutor的使用还是很有技巧的。
    使用无界queue可能会耗尽系统资源。
    使用有界queue可能不能很好的满足性能，需要调节线程数和queue大小
    线程数自然也有开销，所以需要根据不同应用进行调节。

通常来说对于静态任务可以归为：

    数量大，但是执行时间很短
    数量小，但是执行时间较长
    数量又大执行时间又长
    除了以上特点外，任务间还有些内在关系

看完这篇问文章后，希望能够可以选择合适的类型了

参考：http://dongxuan.iteye.com/blog/901689

newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
参考： http://www.trinea.cn/android/java-android-thread-pool/
项目：threadpool-executor

3.execute方法JDK 实现
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
    }
}
一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个Runnable类型的对象，任务的执行方法就是run()方法，如果传入的为null，侧抛出NullPointerException。
如果当前线程数小于corePoolSize，调用addIfUnderCorePoolSize方法，addIfUnderCorePoolSize方法首先调用mainLock加锁，再次判断当前线程数小于corePoolSize并且线程池处于RUNNING状态，则调用addThread增加线程
addIfUnderCorePoolSize方法实现：
    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;
    }


addThread方法首先创建Work对象，然后调用threadFactory创建新的线程，如果创建的线程不为null，将Work对象的thread属性设置为此创建出来的线程，并将此Work对象放入workers中，然后在增加当前线程池的中线程数，增加后回到addIfUnderCorePoolSize方法，释放mainLock，最后启动这个新创建的线程来执行新传入的任务。

addThread方法实现：
private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);<span style="color:#ff0000;"></span>
        if (t != null) {
            w.thread = t;
            workers.add(w);
            int nt = ++poolSize;
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }


ThreadFactory 接口默认实现DefaultThreadFactory
    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;
    }

从addThread方法看得出，Worker对象包装了参数传入的任务，threadFactory新创建的线程包装了Worker对象，在执行新创建线程的run方法时，调用到了Worker对象的run方法.

Worker的run方法
    public void run() {
        try {
            Runnable task = firstTask;
            firstTask = null;
            while (task != null || (task = getTask()) != null) {
                runTask(task);
                task = null;
            }
        } finally {
            workerDone(this);
        }
    }

从以上方法可以看出，Worker所在的线程启动后，首先执行创建其时传入的Runnable任务，执行完成后，循环调用getTask来获取新的任务，在没有任务的情况下，退出此线程。

getTask方法实现：

    Runnable getTask() {
        for (;;) {
            try {
                int state = runState;
                if (state > SHUTDOWN)
                    return null;
                Runnable r;
                if (state == SHUTDOWN) // Help drain queue
                    r = workQueue.poll();
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
                else
                    r = workQueue.take();
                if (r != null)
                    return r;
                if (workerCanExit()) {
                    if (runState >= SHUTDOWN) // Wake up others
                        interruptIdleWorkers();
                    return null;
                }
                // Else retry
            } catch (InterruptedException ie) {
                // On interruption, re-check runState
            }
        }
    }

getTask就是通过WorkQueue的poll或task方法来获取下一个要执行的任务。

回到execute方法，execute 方法部分实现：
    if (runState == RUNNING && workQueue.offer(command)) {
                   if (runState != RUNNING || poolSize == 0)
                       ensureQueuedTaskHandled(command);
               }
               else if (!addIfUnderMaximumPoolSize(command))
                   reject(command); // is shutdown or saturated

如果当前线程池数量大于corePoolSize或addIfUnderCorePoolSize方法执行失败，则执行后续操作；如果线程池处于运行状态并且workQueue中成功加入任务，再次判断如果线程池的状态不为运行状态或当前线程池数为0，则调用ensureQueuedTaskHandled方法

ensureQueuedTaskHandled方法实现：
    private void ensureQueuedTaskHandled(Runnable command) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        boolean reject = false;
        Thread t = null;
        try {
            int state = runState;
            if (state != RUNNING && workQueue.remove(command))
                reject = true;
            else if (state < STOP &&
                     poolSize < Math.max(corePoolSize, 1) &&
                     !workQueue.isEmpty())
                t = addThread(null);
        } finally {
            mainLock.unlock();
        }
        if (reject)
            reject(command);
        else if (t != null)
            t.start();
    }


ensureQueuedTaskHandled方法判断线程池运行，如果状态不为运行状态，从workQueue中删除，并调用reject做拒绝处理。

reject方法实现:

[java] view plain copy
print?

    void reject(Runnable command) {
        handler.rejectedExecution(command, this);
    }

再次回到execute方法，
[java] view plain copy
print?

    if (runState == RUNNING && workQueue.offer(command)) {
                   if (runState != RUNNING || poolSize == 0)
                       ensureQueuedTaskHandled(command);
               }
               else if (!addIfUnderMaximumPoolSize(command))
                   reject(command); // is shutdown or saturated

如线程池workQueue offer失败或不处于运行状态，调用addIfUnderMaximumPoolSize，addIfUnderMaximumPoolSize方法基本和addIfUnderCorePoolSize实现类似，不同点在于根据最大线程数（maximumPoolSize）进行比较，如果超过最大线程数，返回false，调用reject方法，下面是addIfUnderMaximumPoolSize方法实现：

[java] view plain copy
print?

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


3. 添加任务处理流程
当一个任务通过execute(Runnable)方法欲添加到线程池时：
如果当前线程池中的数量小于corePoolSize，并线程池处于Running状态，创建并添加的任务。
如果当前线程池中的数量等于corePoolSize，并线程池处于Running状态，缓冲队列 workQueue未满，那么任务被放入缓冲队列、等待任务调度执行。
如果当前线程池中的数量大于corePoolSize，缓冲队列workQueue已满，并且线程池中的数量小于maximumPoolSize，新提交任务会创建新线程执行任务。

如果当前线程池中的数量大于corePoolSize，缓冲队列workQueue已满，并且线程池中的数量等于maximumPoolSize，新提交任务由Handler处理。

当线程池中的线程大于corePoolSize时，多余线程空闲时间超过keepAliveTime时，会关闭这部分线程。

4. RejectedExecutionHandler 默认有四个选择：

ThreadPoolExecutor.AbortPolicy()              当线程池中的数量等于最大线程数时、直接抛出抛出Java.util.concurrent.RejectedExecutionException异常

[java] view plain copy
print?

    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws 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());
        }
    }

ThreadPoolExecutor.CallerRunsPolicy()       当线程池中的数量等于最大线程数时、重试执行当前的任务，交由调用者线程来执行任务

[java] view plain copy
print?

    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.DiscardOldestPolicy()   当线程池中的数量等于最大线程数时、抛弃线程池中最后一个要执行的任务，并执行新传入的任务

[java] view plain copy
print?

    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.
           *
           * @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()) {
                  e.getQueue().poll();
                  e.execute(r);
              }
          }
      }

ThreadPoolExecutor.DiscardPolicy()            当线程池中的数量等于最大线程数时，不做任何动作
[java] view plain copy
print?

    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) {
        }
    }

分享到：

Ehcache与Guava Cache的区别浅谈 | 池化技术详解-对象池，连接池，线程池

2017-08-12 21:42
浏览 1177
评论(0)
分类:企业架构
查看更多

发表评论

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

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论