`

Java多线程笔记6-ThreadPoolExecutor/ExecutorService

 
阅读更多

1.ThreadPoolExecutor代码示例

package com.landon.mavs.example.concurrent;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.landon.mavs.example.util.MavsCachedThreadPoolExecutor;
import com.landon.mavs.example.util.MavsFixedThreadPoolExecutor;
import com.landon.mavs.example.util.MavsRejectedExecutionPolicy;
import com.landon.mavs.example.util.MavsThreadFactory;
import com.landon.mavs.example.util.MavsThreadPoolStateMonitor;

/**
 * 
 * {@link java.util.concurrent.ThreadPoolExecutor}示例
 * 
 * 

 * 1.public class ThreadPoolExecutor extends AbstractExecutorService
 * 
 * 2.AbstractExecutorService内部的提交任务方法系列最终均调用了execute方法执行任务.{@link java.util.concurrent.RunnableFuture}
 * 
 * 3.Executors:
 *       // 1.corePoolSize和maximumPoolSize一致.
 *       // 2.keepAliveTime传入0L,即在队列无元素时则直接不等待直接返回null(因线程池大小不会超过corePoolSize_无界阻塞队列且【1描述】).
 *                因该参数是在getTask方法当(poolSize > corePoolSize || allowCoreThreadTimeOut)时调用,而第一个条件已经是false了,即当队列空的时候
 *             则一直会阻塞(take).所以只有在线程池设置了allowCoreThreadTimeOut参数时才会进行调用.而在allowCoreThreadTimeOut(boolean value)方法
 *             的实现中,如果(value && keepAliveTime <= 0)则抛出异常.即allowCoreThreadTimeOut(true)和keepAliveTime<=0这两个参数不能同时存在.
 *             所以在FixedThreadPool实现中keepAliveTime参数无效(即永远不会回收Worker线程).
 *       // 3.workQueue为LinkedBlockingQueue(未指定capacity),即无界阻塞队列.则线程池大小>=corePoolSize时则将任务插入队列.
 *       // 4.总结:FixedThreadPool正如其名字一样,线程池中的线程数目是Fixed,固定的,Worker线程不会被回收且队列无任务时则一直阻塞.
 *       public static ExecutorService newFixedThreadPool(int nThreads) {
 *               return new ThreadPoolExecutor(nThreads, nThreads,
 *                                             0L, TimeUnit.MILLISECONDS,
 *                                             new LinkedBlockingQueue());
 *           }
 * 
 *      // 单线程+无界阻塞消息队列的经典模型.无线程安全问题.
 *  // 注意其返回的是封装的FinalizableDelegatedExecutorService并实现了finalize方法,而finalize方法则调用了线程池的shutdown方法.
 *  // 同时要主要强转的问题.其实际类型不是ThreadPoolExecutor
 *  // 而FinalizableDelegatedExecutorService继承了DelegatedExecutorService(委托 /代理),其只是包装了,仅暴露了ExecutorService的实现方法.
 *  // 个人认为因为其就是是单线程的,所以完全没有必要暴露ThreadPoolExecutor的所有访问方法,暴露了反而可能因为不必要的麻烦.
 *  // 但是之所以再封装一层finalize就不知为何了.(GC回收之前的调用?有啥必要呢?如果我们没有对它调用shutdown(),那么可以确保它在被回收时调用shutdown()来终止线程)
 *  // 所以只能用安全网来解释这个设计了
 *  // landon:终于明白了.1.其主要原因只是要暴露ExecutorService的方法,不要暴露ThreadPoolExecutor的所有访问方法
 *                    2.加上finalize的原因在于ThreadPoolExecutor本身有finalize方法,且实现为shutdown.而DelegatedExecutorService本身是没有的.
 *      所以额外加了在FinalizableDelegatedExecutorService加上了finalize.与ThreadPoolExecutor的finalize保持一致.
 *      public static ExecutorService newSingleThreadExecutor() {
 *         return new FinalizableDelegatedExecutorService
 *             (new ThreadPoolExecutor(1, 1,
 *                                     0L, TimeUnit.MILLISECONDS,
 *                                     new LinkedBlockingQueue()));
 *     }
 *     
 *   // 1.corePoolSize为0,maximumPoolSize为Integer.MAX_VALUE.则执行任务的时候会直接向workQueue.offer任务.
 *   // 2.workQueue为SynchronousQueue,即同步阻塞队列(非公平),即offer的是恰好有线程poll才可以成功.第一次执行任务的时候,offer肯定fail.所以
 *           ->addIfUnderMaximumPoolSize->即添加一个worker线程.(常规情况下会一直UnderMaximum.因为Integer.MAX_VALUE)
 *   // 3.keepAliveTime为60秒.而poolSize一定大于corePoolSize(为0)->workQueue.poll(keepAliveTime)->即从工作队列poll.所以说
 *           如果在60秒内有任务offer则worker线程getTask成功则执行任务;否则返回null,又因为是同步阻塞队列所有在判断workerCanExit的时候(isEmpty永远为true),
 *           所以worker线程会退出被回收.
 *   // 总结:1.如果线程池繁忙的情况下,每个线程都在执行任务的时候,新任务会新建新的worker线程去执行任务.
 *           2.假如在提交新任务的时候恰好有线程正在空闲getTask(60s超时内)则会委托空闲线程去做.
 *           3.如果线程池不繁忙,偶尔来一个任务.则第一个任务会创建一个workder线程,此时执行完毕如果1分钟内还没有任务则该线程会被自动回收.即该线程池最小的线程数目其实是0.
 *   public static ExecutorService newCachedThreadPool() {
 *         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
 *                                       60L, TimeUnit.SECONDS,
 *                                       new SynchronousQueue());
 *     }
 *     
 * 4.RejectPolicy
 *     1.ThreadPoolExecutor内部预定义了4中拒绝的处理程序策略
 *  2.Reject的执行时机:
 *      1.当提交一个任务t的时候,线程池中数目超出了coreSize->提交至workQueue.此时检查线程池状态不在是running或线程池中突然没有了线程(有可能是额外线程调用了线程池
 *          的shutdown/shutdownNow)->ensureQueuedTaskHandled->即如果发现此时状态不是running且可以从workQueue将t移除,则执行拒绝策略.即shutdown的时候
 *          新的任务会被拒绝.->后续如果发现是调用了shutdown(线程池状态是SHUTDOWN)->且workQueue不为空且poolSize添加worker线程->即保证
 *          队列任务执行完毕.{@link #shutdown}方法会中断空闲的线程(超出coreSize的线程) {@link getTask}
 *         2.当线程池线程数目超出maximumPoolSize的时候则执行拒绝策略.
 *     3.
 *         1.AbortPolicy:终止策略
 *             // 其是线程池的默认拒绝策略defaultHandler.->execute方法的调用线程则直接抛出异常
 *             public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 *             throw new RejectedExecutionException();
 *         }
 *         
 *      2.CallerRunsPolicy:调用者运行策略
 *          // 直接在execute方法的调用线程运行被拒绝的任务.如果线程池已关闭则丢弃该任务.
 *          // 因为是在execute的调用线程中运行的.所以可简单的减缓新任务的提交速度.即得等到执行完被线程池拒绝的任务后才能提交任务.
 *          public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 *             if (!e.isShutdown()) {
 *                 r.run();
 *             }
 *         }
 *         
 *       3.DiscardOldestPolicy:丢弃最旧的任务策略
 *           // 放弃最旧的未处理的任务(即队头元素),重新提交执行被拒绝的任务r.如果线程池已关闭则丢弃该任务.
 *           public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 *             if (!e.isShutdown()) {
 *                 e.getQueue().poll();
 *                 e.execute(r);
 *             }
 *         }
 *         
 *       4.DiscardPolicy:丢弃策略
 *           // 空实现,即直接丢弃被拒绝的任务
 *           public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 *         }
 *         
 *   5.钩子方法:
 *       1.protected void beforeExecute(Thread t, Runnable r),可在子类覆写.在执行的线程运行任务之前调用的方法.此方法由t调用.
 *           ->(方法结束时,子类通常应该调用super.beforeExecute->嵌套多个重写操作)
 *      2.protected void afterExecute(Runnable r, Throwable t) ,可在子类覆写,完成给定任务后所调用的方法.此方法由执行任务的worker线程
 *          调用.t为执行该任务时导致终止时的异常->该异常会被抛到上层run然后被try.(没有catch).->如果t为null,则表示任务执行顺利.
 *          注:当提交的任务类似于submit方法提交的(如FutureTask时)->{@link FutureTask$Sync#innerRun}会在内部捕获该异常.所以其不会导致worker
 *          线程突然终止.而异常也不会传递给该方法。
 *          ->(方法开始时,子类通常应该调用super.afterExecute->嵌套多个重写操作)
 *          注:workerDone方法是在worker线程结束时调用的方法->完成任务计数
 *      3.protected void terminated,此为线程池终止时调用的方法{@link #tryTerminate}.子类可覆写.
 *  
 *  6.public BlockingQueue getQueue(),该方法用来访问工作队列.->用于监控和调试目的.->强烈反对出于其他目的而使用此方法.
 *  
 *  7.public boolean remove(Runnable task),从线程池的内部工作队列中移除此任务.如果其尚未开始,则其不再运行.注:对于通过submit输入的runnable无法移除.
 *      因为其已经被转换了其他形式如FutureTask.
 *  
 *  8. public void purge(),尝试从工作队列移除已取消的Future任务.->取消的任务不会再次执行.但是他们可能在工作队列中累积.直到worker线程将其主动移除(从工作队列poll).
 *      该方法则试图移除他们.如果出现其他线程的干预->则抛出ConcurrentModificationException.则失败.
 *  
 *  9. 即使用户忘记调用了shutdown关闭线程池:也希望确保可回收线程->设置keepAliveTime/allowCoreThreadTimeOut/corePoolSize为0.
 *      {@link #getTask()}
 *  
 *  10.public boolean prestartCoreThread() 启动核心线程,使其处于getTask的空闲状态. 如果已启动了线程,则返回true
 *      从源码上:其内部直接调用了addIfUnderCorePoolSize(null).既如果coreSize不为0,则会启动一个worker线程并处于getTask的等待状态
 *  
 *  11.public int prestartAllCoreThreads() 启动所有核心线程,使其处于等待任务的空闲状态
 *      从源码上: while (addIfUnderCorePoolSize(null))->即超出coreSize则跳出循环. 返回已启动的线程数
 *  
 *  12.Worker#isActive
 *      // runLock是在runTask方法内的锁.而不是run的锁.即如果线程在getTask等待空闲的时候不是active.只有在真正执行任务的时候是active.
 *      boolean isActive() {
 *             return runLock.isLocked();
 *         }
 *   
 *  13.public void setCorePoolSize(int corePoolSize) 设置核心线程数
 *      从源码上:1.设置coreSize为传入的新值
 *            2.如果新值大于旧值,则会添加额外线程,但是启动的线程数目一定不会超过当前工作队列的大小.
 *            3.如果新值小于旧值,则会遍历当前所有的worker线程,将多出的线程进行interruptIfIdle().其中还有一个条件是workQueue.remainingCapacity() == 0.
 *              也就是说要求此时工作队列的可附加元素数量为0,则当前工作队列已满.
 *              (个人认为这个条件的添加是因为如果此时工作队列已满,则再次提交任务的时候会在maximum之下继续添加线程的.也就是说在这时候中断一个core线程是没有问题的.)
 *          ->如果条件不满足的时候则多余的现有线程将在下一次空闲时终止(因为poolSize > coreSize).{@link #getTask}
 *  
 *  14.public void setMaximumPoolSize(int maximumPoolSize) 设置运行最大的线程数
 *      从源码上:1.参数maximumPoolSize必须>0 且 >=corePoolSize,否则抛出IllegalArgumentException.
 *            2.设置新值.
 *            3.如果新值小于当前值且当前poolSize > maximumPoolSize->则会遍历工作线程,将多余的线程interruptIfIdle.
 *  
 *  15.线程池在调用了shutdown方法后便不能在提交任务了,因为此时的线程池状态已经不是running了.但是如果线程因为执行任务而异常终止的话,却依然可以提交任务.
 *      因为此时状态还是running.
 * 

 * 
 * @author landon
 * 
 
*/

public class ThreadPoolExecutorExample {
    
private static final Logger LOGGER = LoggerFactory
            .getLogger(ThreadPoolExecutorExample.class);

    
public static void main(String[] args) throws Exception {
        
// 固定2个线程的线程池
        MavsFixedThreadPoolExecutor fixedThreadPoolExecutor1 = new MavsFixedThreadPoolExecutor(
                
2new MavsThreadFactory("Example""FixedThreadPool-1"),
                
new MavsRejectedExecutionPolicy());
        
// 从线程池的状态监视器来看:此时poolSize=1/workQueueSize=0,即启动了一个线程,工作队列没有任务
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());
        
// 从线程池的状态监视器来看:此时poolSize=2/workQueueSize=0,即又启动了一个线程
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());

        
// 由提交了3个任务,从输出来看:poolSize一直为2.而workQueueSize最多为3->随着任务的执行,workQueueSize变为0
        
// 所以MavsFixedThreadPoolExecutor这个线程池会保持固定线程数量
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());

        
// 执行shutdown
        
// 另外从输出看:发现线程池终止的时候调用了terminate方法
        fixedThreadPoolExecutor1.shutdown();

        Thread.sleep(1 * 1000);

        
// 测试shutdown后,还可以执行任务吗?
        
// 答案当然是NO.因为新建worker线程的条件包括插入队列都必须是在RUNNING状态下.
        
// 而执行了shutdown后则更改了运行状态为SHUTDOWN
        fixedThreadPoolExecutor1.execute(new ThreadPoolTask());

        Thread.sleep(1 * 1000);
        LOGGER.debug("");

        
// cached线程池
        MavsCachedThreadPoolExecutor cachedThreadPoolExecutor1 = new MavsCachedThreadPoolExecutor(
                
new MavsThreadFactory("Example""CachedThreadPool-1"),
                
new MavsRejectedExecutionPolicy());

        
// 从输出可以看到,线程池最多启动了5个线程,workQueueSize一直为0
        cachedThreadPoolExecutor1.execute(new ThreadPoolTask());
        cachedThreadPoolExecutor1.execute(new ThreadPoolTask());
        cachedThreadPoolExecutor1.execute(new ThreadPoolTask());
        cachedThreadPoolExecutor1.execute(new ThreadPoolTask());
        cachedThreadPoolExecutor1.execute(new ThreadPoolTask());

        
// 暂停2分钟,使得默认空闲1分钟的worker线程退出
        Thread.sleep(2 * 60 * 1000);

        LOGGER.debug("");

        
// 从输出可以看到:poolSize=0,即空闲的worker线程被回收了.
        
// 另外所有的worker线程被回收了,线程池就结束了.
        
// 因为:ThreadPoolExecutor#void workerDone(Worker w)->
        
// if (--poolSize ==0)tryTerminate()
        
// 但是这种线程自然结束的话,并没有调用覆写的terminate方法.因为tryTerminate的实现中是判断当前线程池状态是STOP/SHUTDOWN的时候才执行terminated方法的
        LOGGER.debug("cachedThreadPoolExecutor1.state:{}",
                MavsThreadPoolStateMonitor.monitor(cachedThreadPoolExecutor1));

        
// 单线程线程池,注意这个和{@link
        
// Executors#newSingleThreadExecutor的区别},后者仅是返回的暴露的ExecutorService接口
        MavsFixedThreadPoolExecutor singleExecutor = new MavsFixedThreadPoolExecutor(
                
1new MavsThreadFactory("Example""SingleThreadPool-1"),
                
new MavsRejectedExecutionPolicy());
        
// 提交一个可抛出异常的任务
        
// 从输出看出
        
// 1:执行了afterExecute方法且其中的Throwable t为不null.此执行任务的时候抛出了异常.
        
// 2.线程因为异常终止,因指定了线程默认的UncaughtExceptionHandler,所以执行了uncaughtException方法.
        singleExecutor.execute(new ThreadPoolExceptionTask());
        Thread.sleep(1 * 1000);
        
// 从输出可以看到:poolSize=0变为了0.即线程终止了.
        
// 因为Worker线程的run方法只是try/finally,即并没有捕获异常.而runTask向上抛出异常至run,直接到finally.->workerDone->poolSize--
        
// ->tryTerminate
        LOGGER.debug("singleExecutor.state:{}",
                MavsThreadPoolStateMonitor.monitor(singleExecutor));

        Thread.sleep(1 * 1000);
        
// 测试线程池异常终止后,还可以执行任务吗?
        
// 答案是YES.因为此时的线程池状态依然是RUNNING.
        singleExecutor.execute(new ThreadPoolTask());
        Thread.sleep(1 * 1000);
        
// 从输出发现:poolSize=1,即新增了一个worker线程.另外从线程的名字Mavs-Example-SingleThreadPool-1-2也可看得出.
        LOGGER.debug("singleExecutor.state:{}",
                MavsThreadPoolStateMonitor.monitor(singleExecutor));

        
// 这里是提交了一个任务,内部会被封装成->RunnableFuture->FutureTask
        
// 而其内部run->Sync#innerRun->其内部会被try/catch的->所以理论上结果应该线程应该不会异常终止.
        
// 从输出看:1.afterExecute方法中的异常参数为null.
        
// 2.没用调用默认的UncaughtExceptionHandler.也就是说线程正常运行.
        singleExecutor.submit(new ThreadPoolExceptionTask());

        singleExecutor.shutdown();

        
// 测试setCoreSize以及setMaximumSize

        
// 3个固定线程数目的线程池
        MavsFixedThreadPoolExecutor fixedThreadPoolExecutor2 = new MavsFixedThreadPoolExecutor(
                
3new MavsThreadFactory("Example""FixedThreadPool-2"),
                
new MavsRejectedExecutionPolicy());

        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());

        
// 设置核心线程大小为6.
        fixedThreadPoolExecutor2.setCorePoolSize(6);

        Thread.sleep(1 * 1000);
        
// 从输出看:poolSize=6
        LOGGER.debug("fixedThreadPoolExecutor2.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor2));

        
// 设置核心线程大小为2
        fixedThreadPoolExecutor2.setCorePoolSize(2);
        Thread.sleep(1 * 1000);
        
// 从输出看.poolSize=6
        
// 因为 workQueue.remainingCapacity()此时不为0,即不会中断多余的空闲线程.
        
// 另外此时所有的worker线程正在处在等待状态.
        LOGGER.debug("fixedThreadPoolExecutor2.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor2));

        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        Thread.sleep(1 * 1000);
        
// 从输出可以看到:此时poolSize=5.因为某个等待线程获得执行机会后再次getTask后->会执行pool(keepAliveTime),则直接回收退出.
        LOGGER.debug("fixedThreadPoolExecutor2.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor2));

        
// 继续执行3个任务
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor2.execute(new ThreadPoolTask());
        Thread.sleep(1 * 1000);
        
// 从输出可以发现:此时poolSize=2,因为多余的线程在执行完任务下次getTask判断的时候直接就被回收了.
        
// 另外:此时maximumSize是3.coreSize为2.也就是说此时的线程池已经不再是固定数量线程的线程池了.
        LOGGER.debug("fixedThreadPoolExecutor2.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor2));

        fixedThreadPoolExecutor2.shutdown();

        
// 测试setMaximumPoolSize
        
// 2个固定线程数目的线程池
        MavsFixedThreadPoolExecutor fixedThreadPoolExecutor3 = new MavsFixedThreadPoolExecutor(
                
2new MavsThreadFactory("Example""FixedThreadPool-3"),
                
new MavsRejectedExecutionPolicy());

        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());

        Thread.sleep(1 * 1000);
        
// 设置最大线程池大小为4
        fixedThreadPoolExecutor3.setMaximumPoolSize(4);

        
// 提交一系列任务
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());
        fixedThreadPoolExecutor3.execute(new ThreadPoolTask());

        Thread.sleep(1 * 1000);
        
// 从输出看:maximumPoolSize=4/poolSize=2
        
// 即只是修改了maximumPoolSize的值/poolSize仍然为2.因为用的是无限阻塞队列,所以多余的任务都被放到了队列.
        LOGGER.debug("fixedThreadPoolExecutor3.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor3));
        
try {
            
// 这里抛出了一个异常,因为1比coreSize 2还要小
            fixedThreadPoolExecutor3.setMaximumPoolSize(1);
        }
 catch (Exception e) {
            LOGGER.warn("fixedThreadPoolExecutor3.setMaximumPoolSize.err.", e);
        }


        fixedThreadPoolExecutor3.shutdown();

        
// 自定义线程池1
        
// 工作队列为容量3的阻塞队列
        
// 等待空闲时间为60s
        ThreadPoolExecutor userDefinedExecutor1 = new ThreadPoolExecutor(24,
                
10, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3),
                
new MavsThreadFactory("Example""User-Define-Executor-1"),
                
new MavsRejectedExecutionPolicy());

        
// 直接提交很多任务
        
// 这个测试的目的在于测试拒绝策略.从输出可以看到:
        
// poolSize=4/workQueueSize=3这个时候,即已经达到了最大线程数目和队列上限,则执行了拒绝策略.
        for (int i = 0; i < 20; i++{
            userDefinedExecutor1.execute(new ThreadPoolTask());
        }


        Thread.sleep(5 * 1000);
        LOGGER.debug("userDefinedExecutor1.state:{}",
                MavsThreadPoolStateMonitor.monitor(userDefinedExecutor1));
        
// 将线程池最大池数目调整为3.此时的poolSize为4.
        userDefinedExecutor1.setMaximumPoolSize(3);
        Thread.sleep(1 * 1000);
        
// 从输出看:poolSize还是为4.因为此时所有的worker线程都在poll(timeout)->然后setMaximumPoolSize->会中断一个空闲线程->但是getTask这里
        
// 被try/catch了.
        
// 不过多余的线程在空闲的时候都会被回收.
        LOGGER.debug("userDefinedExecutor1.state:{}",
                MavsThreadPoolStateMonitor.monitor(userDefinedExecutor1));
        Thread.sleep(5 * 1000);
        LOGGER.debug("userDefinedExecutor1.state:{}",
                MavsThreadPoolStateMonitor.monitor(userDefinedExecutor1));

        userDefinedExecutor1.shutdown();

        
// 测试prestartCoreThread()/prestartAllCoreThreads
        MavsFixedThreadPoolExecutor fixedThreadPoolExecutor4 = new MavsFixedThreadPoolExecutor(
                
3new MavsThreadFactory("Example""FixedThreadPool-4"),
                
new MavsRejectedExecutionPolicy());

        LOGGER.debug("fixedThreadPoolExecutor4.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor4));
        
// 启动一个核心线程
        fixedThreadPoolExecutor4.prestartCoreThread();
        
// 从输出可以看出:poolSize为1,即启动了一个worker.
        LOGGER.debug("fixedThreadPoolExecutor4.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor4));
        
// 启动所有核心线程
        
// 从输出可以看出:poolSize为3,即现在启动了所有的核心线程
        fixedThreadPoolExecutor4.prestartAllCoreThreads();
        LOGGER.debug("fixedThreadPoolExecutor4.state:{}",
                MavsThreadPoolStateMonitor.monitor(fixedThreadPoolExecutor4));
    }


    
/**
     * 
     * 用于测试的线程池任务
     * 
     * @author landon
     * 
     
*/

    
private static class ThreadPoolTask implements Runnable {
        
private static final AtomicInteger COUNTER = new AtomicInteger(1);

        
private int id;

        
public ThreadPoolTask() {
            id = COUNTER.getAndIncrement();
        }


        @Override
        
public void run() {
            LOGGER.debug(this + " begin");

            
try {
                TimeUnit.MICROSECONDS.sleep(100);
            }
 catch (InterruptedException e) {
                LOGGER.warn(this + " was interrupted", e);
            }


            LOGGER.debug(this + " end");
        }


        @Override
        
public String toString() {
            
return "ThreadPoolTask [id=" + id + "]" + "["
                    
+ Thread.currentThread().getName() + "]";
        }

    }


    
/**
     * 
     * 用于测试的线程池异常任务
     * 
     * @author landon
     * 
     
*/

    
private static class ThreadPoolExceptionTask implements Runnable {
        
private static final AtomicInteger COUNTER = new AtomicInteger(1);

        
private int id;

        
public ThreadPoolExceptionTask() {
            id = COUNTER.getAndIncrement();
        }


        @Override
        
public void run() {
            LOGGER.debug(this + " begin");

            
throw new RuntimeException("ThreadPoolExceptionTask.Exception.");
        }


        @Override
        
public String toString() {
            
return "ThreadPoolExceptionTask [id=" + id + "]" + "["
                    
+ Thread.currentThread().getName() + "]";
        }

    }

}



2.ExecutorService示例代码


package com.landon.mavs.example.concurrent;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
* ExecutorServiceExample
*
* @author landon
*
*/

publicclass ExecutorServiceExample {
   
privatestaticfinal Logger LOGGER = LoggerFactory
            .getLogger(ExecutorServiceExample.class);

   
publicstaticvoid main(String[] args) {
        ExecutorService exeSrv = Executors.newFixedThreadPool(4);
       
// execute(Runnable command) 执行一个Runnable
        exeSrv.execute(new OneRunnable(1));

       
// Future submit(Runnable task) 提交一个Runable
        Future oneRunFuture = exeSrv.submit(new OneRunnable(2));
       
// Future#isDone 返回任务是否结束
        LOGGER.debug("oneRun is complete:"+ oneRunFuture.isDone());
       
try{
           
// 等待计算完成,返回计算结果
           
// 当前成功完成的时候 #get 返回null
            LOGGER.debug("oneRun result:"+ oneRunFuture.get());
        }
catch (InterruptedException e) {
            LOGGER.warn("exception_oneRun#get is interrupted while waiting result.");
        }
catch (ExecutionException e) {
            LOGGER.warn("exception_oneRun#get compuation throws a exception");
        }


       
// Future submit(Callable task) 提交一个Callable
        Future<String> oneCallFuture = exeSrv.submit(new OneCallable(1));

       
try{
           
// V get() throws InterruptedException, ExecutionException
           
// 等待计算完成,返回计算结果
            LOGGER.debug("oneCall result:"+ oneCallFuture.get());
        }
catch (InterruptedException e) {
            LOGGER.warn("exception_oneCall#get is interrupted while waiting result.");
        }
catch (ExecutionException e) {
            LOGGER.warn("exception_oneCall#get compuation throws a exception");
        }


        Future<String> oneCallFuture2 = exeSrv.submit(new OneCallable(2));

       
try{
           
// V get(long timeout, TimeUnit unit) 指定等待超时时间
            LOGGER.debug("oneCall2 result:"
                   
+ oneCallFuture2.get(1, TimeUnit.SECONDS));
        }
catch (InterruptedException e) {
            LOGGER.warn("exception_oneCall2#get is interrupted while waiting result.");
        }
catch (ExecutionException e) {
            LOGGER.warn("exception_oneCall2#get compuation throws a exception");
        }
catch (TimeoutException e) {
            LOGGER.warn("exception_oneCall2#get timeout");
        }


        Future<String> oneCallFuture3 = exeSrv.submit(new OneCallable(3));
       
// boolean cancel(boolean mayInterruptIfRunning)
       
// 尝试取消任务的执行.如果任务已完成或者已经被取消或者因为其他一些原因被能取消则尝试会失败
       
// 如果尝试成功且任务还未开始则该任务再也不会运行.如果任务已经启动,mayInterruptIfRunning参数决定任务执行线程是否中断尝试结束任务
        boolean isFuture3CancelSuccess = oneCallFuture3.cancel(false);
        LOGGER.debug("oneCallFuture3#cancel(false) result:"
               
+ isFuture3CancelSuccess);
        LOGGER.debug("oneCallFuture3#isDone:"+ oneCallFuture3.isDone());
        LOGGER.debug("oneCallFuture3#isCancelled:"
               
+ oneCallFuture3.isCancelled());

        Future<String> oneCallFuture4 = exeSrv.submit(new OneCallable(4));
       
// 主线程暂停2秒后执行cancel
        try{
            TimeUnit.SECONDS.sleep(2);
        }
catch (InterruptedException e) {

        }

       
// 此处cancel传true则表明如果任务已启动则中断执行任务线程尝试结束任务
       
// 从输出可以看到,即输出了任务开始,但是却没有输出任务结束->且返回true表明任务被中断取消
        boolean isFuture4CancelSuccess = oneCallFuture4.cancel(true);
        LOGGER.debug("oneCallFuture4#cancel(true) result:"
               
+ isFuture4CancelSuccess);
        LOGGER.debug("oneCallFuture4#isDone:"+ oneCallFuture4.isDone());
        LOGGER.debug("oneCallFuture4#isCancelled:"
               
+ oneCallFuture4.isCancelled());

        Future<String> oneCallFuture5 = exeSrv.submit(new OneCallable(5));
       
// 主线程暂停8秒后执行cancel,此时任务有可能已经执行完毕
        try{
            TimeUnit.SECONDS.sleep(8);
        }
catch (InterruptedException e) {

        }

       
// 从输入可以看到,任务5输出了end.即cancel时任务已经完成.所以isFuture5CancelSuccess为false.isDone为true.isCancelled为false
        boolean isFuture5CancelSuccess = oneCallFuture5.cancel(true);
        LOGGER.debug("oneCallFuture5#cancel(true) result:"
               
+ isFuture5CancelSuccess);
        LOGGER.debug("oneCallFuture5#isDone:"+ oneCallFuture5.isDone());
        LOGGER.debug("oneCallFuture5#isCancelled:"
               
+ oneCallFuture5.isCancelled());

       
// Future submit(Runnable task, T result) 当任务完成时get方法会返回指定的result
        Future<String> oneRun3Future = exeSrv
                .submit(new OneRunnable(3), "isOk");
       
try{
           
// 从输入可以看到get方法的返回是传入的"isOk"
            LOGGER.debug("oneRun3 result:"+ oneRun3Future.get());
        }
catch (InterruptedException e) {
            LOGGER.warn("exception_oneRun3l#get is interrupted while waiting result.");
        }
catch (ExecutionException e) {
            LOGGER.warn("exception_oneRun3#get compuation throws a exception");
        }


       
// 任务集合
        List<OneCallable> oneCallList = Arrays.asList(new OneCallable(10),
               
new OneCallable(11), new OneCallable(12));
       
try{
           
// List> invokeAll(Collection>
           
// tasks) throws InterruptedException;
           
// 相当于批量执行任务.从方法的异常列表可以看出.此方法会等待(即阻塞)直到所有任务完成
            List<Future<String>> oneCallListFutures = exeSrv
                    .invokeAll(oneCallList);

           
// 处理完成结果 从输出可以看到->invokeAll确实是在等待所有任务执行完毕.
            List<Boolean> resultList =new ArrayList<Boolean>();
           
for (Iterator<Future<String>> iterator = oneCallListFutures
                    .iterator(); iterator.hasNext();) {
               
if (iterator.next().isDone()) {
                    resultList.add(true);
                }

            }


            LOGGER.debug("oneCallListFutures result: "+ resultList);
        }
catch (InterruptedException e) {
            LOGGER.warn("exeSrv#invokeAll(oneCallList) exception_waiting all task complete was interrupted.");
        }


       
// 任务集合2
        List<OneCallable> oneCallList2 = Arrays.asList(new OneCallable(20),
               
new OneCallable(21), new OneCallable(22));
       
try{
           
// T invokeAny(Collection> tasks) throws
           
// InterruptedException, ExecutionException;
           
// 批量执行任务->等待直到某个任务已成功完成(注意只要某个任务成功返回则返回结果) 另外注意返回结果是T,而非Future
            String oneCallList2Result = exeSrv.invokeAny(oneCallList2);
           
// 从输出结果可以看到:
           
// [oneCallList2Result:OneCallable [taskNum=20]OK],即20号任务执行完成即返回了
            LOGGER.debug("oneCallList2Result:"+ oneCallList2Result);
        }
catch (InterruptedException e) {
            LOGGER.warn("exeSrv#invokeAny(oneCallList2) exception_waiting one task complete was interrupted.");
        }
catch (ExecutionException e) {
            LOGGER.warn("exeSrv#invokeAll(oneCallList2) exception_one any one task was completed.");
        }


       
// 任务集合3
        List<OneCallable> oneCallList3 = Arrays.asList(new OneCallable(30),
               
new OneCallable(31), new OneCallable(32));
       
try{
           
// List> invokeAll(Collection>
           
// tasks, long timeout, TimeUnit unit) throws InterruptedException;
           
// 批量执行任务->指定等待超时时间->注意该方法并不会抛出超时异常,即如果没有被打断的情况下,超时后(直接返回),则某些任务只是未完成而已(注返回后会取消尚未完成的任务)
            List<Future<String>> oneCallList3Futures = exeSrv.invokeAll(
                    oneCallList3, 2, TimeUnit.SECONDS);

            List<Boolean> oneCallList3Results =new ArrayList<Boolean>();
           
for (Future<String> future : oneCallList3Futures) {
               
if (future.isDone()) {
                    oneCallList3Results.add(true);
                }
else{
                    oneCallList3Results.add(false);
                }

            }


           
// 从输出看,这个方法很特殊.即返回的Future列表的isDone方法都返回true.且所有的任务没有输出结束end.
           
// 从API看,即当所有任务完成或者超时(无论哪个首先发生)则返回的Future列表的isDone方法返回true
           
// 一旦返回后,即取消尚未完成的任务
            LOGGER.debug("oneCallList3Results:"+ oneCallList3Results);
        }
catch (InterruptedException e) {
            LOGGER.warn("exeSrv#invokeAll(oneCallList3, 2, TimeUnit.SECONDS) exception_waiting all task compelte was interrupted.");
        }


       
// 任务集合4
        List<OneCallable> oneCallList4 = Arrays.asList(new OneCallable(40),
               
new OneCallable(41), new OneCallable(42));
       
try{
           
// T invokeAny(Collection> tasks,long
           
// timeout, TimeUnit unit) throws InterruptedException,
           
// ExecutionException, TimeoutException;
           
// 批量执行任务->指定等待超时时间->注意这个方法抛出了TimeoutException->即在等待超时后会抛出异常
           
// 从输出结果可以看到,在等待超时后->尚未完成的任务都被取消了,因为输出只有begin没有end
            exeSrv.invokeAny(oneCallList4, 1, TimeUnit.SECONDS);
        }
catch (InterruptedException e) {
            LOGGER.debug("exeSrv#invokeAny(oneCallList4, 1, TimeUnit.SECONDS) exception_waiting any one task complete was interrupted.");
        }
catch (ExecutionException e) {
            LOGGER.debug("exeSrv#invokeAny(oneCallList4, 1, TimeUnit.SECONDS) exception_no any one task was completed");
        }
catch (TimeoutException e) {
            LOGGER.debug("exeSrv#invokeAny(oneCallList4, 1, TimeUnit.SECONDS) exception_waiting timeout");
        }


       
// void shutdown() 启动一次顺序关闭,执行以前提交的任务,但是不接受新任务.如果已经关闭,则调用没有其他作用
        exeSrv.shutdown();
       
// boolean isShutdown()
       
// ThreadPoolExecutor#isShutdown{return runState != RUNNING}
        LOGGER.debug("exeSrv#shutdown.isShutdown:"+ exeSrv.isShutdown());
       
// isTerminated
       
// ThreadPoolExecutor#isTerminated{return runState == TERMINATED}
       
// 如果关闭后所有任务都完成,则返回true.注:必须要先调用shutdown/shutdownNow
       
// 该方法可结合awaitTermination使用awaitTermination,即if(!isTerminated){awaitTermination}
        LOGGER.debug("exeSrv#shutdown.isTerminated:"+ exeSrv.isTerminated());

        ExecutorService exeSrv2 = Executors.newFixedThreadPool(2);

        exeSrv2.submit(new OneRunnable(50));
        exeSrv2.submit(new OneCallable(60));

       
// List shutdownNow()
       
// 试图终止所有正在执行的活动任务.暂停处理正在等待的任务,并返回等待执行的任务列表
       
// 无法保证能够停止正在处理的活动执行任务,但是会尽力尝试.如通过Thread.interrupt这种典型的实现来取消->所以任何任务无法响应中断都可能永远无法停止
       
// 从输出可以看到50号任务被interrupt了(异常被捕获了).而60号的任务其实也被interrupt了,但是异常被抛出到了上层.
        exeSrv2.shutdownNow();
        LOGGER.debug("exeSrv2#shutdown.isShutdown:"+ exeSrv2.isShutdown());
        LOGGER.debug("exeSrv2#shutdown.isTerminated:"+ exeSrv2.isTerminated());

        ExecutorService exeSrv3 = Executors.newFixedThreadPool(2);
        exeSrv3.submit(new OneRunnable(70));
        exeSrv3.submit(new OneRunnable(80));

        exeSrv3.shutdown();
       
// boolean awaitTermination(long timeout, TimeUnit unit) throws
       
// InterruptedException
       
// 1.阻塞直到shutdown请求后,所有任务完成 2.阻塞直到超时 3.阻塞直到当前线程被中断
        try{
           
if (!exeSrv3.isTerminated()) {
                exeSrv3.awaitTermination(10, TimeUnit.SECONDS);
               
// 从输出看.任务花费了5秒即执行完毕(多线程并行).所以线程池所有任务任务完成后,awaitTermination也不在阻塞.
                LOGGER.debug("exeSrv3EwaitTermination(10, TimeUnit.SECONDS) end.");
            }


        }
catch (InterruptedException e) {
            LOGGER.debug("exeSrv3.awaitTermination(10, TimeUnit.SECONDS) was interrupted.");
        }

    }


   
privatestaticclass OneRunnable implements Runnable {
       
privateint taskNum;

       
public OneRunnable(int taskNum) {
           
this.taskNum = taskNum;
        }


        @Override
       
publicvoid run() {
            LOGGER.debug(this+" begin");

           
// 用sleep模拟业务逻辑耗时
            try{
                TimeUnit.SECONDS.sleep(5);
            }
catch (InterruptedException e) {
                LOGGER.warn("execute"+this+" was interrupt");
            }


            LOGGER.debug(this+" end");
        }


        @Override
       
public String toString() {
           
return"OneRunnable [taskNum="+ taskNum +"]";
        }

    }


   
privatestaticclass OneCallable implements Callable<String>{
       
privateint taskNum;

       
public OneCallable(int taskNum) {
           
this.taskNum = taskNum;
        }


        @Override
       
public String call() throws Exception {
            LOGGER.debug(this+" begin");

           
// 用sleep模拟业务逻辑耗时
            Thread.sleep(3*1000);

            LOGGER.debug(this+" end");

           
returnthis+" OK";
        }


        @Override
       
public String toString() {
           
return"OneCallable [taskNum="+ taskNum +"]";
        }

    }

}



3.附:MavsCachedThreadPoolExecutor/MavsFixedThreadPoolExecutor/MavsRejectedExecutionPolicy/MavsThreadDefaultUncaughtExceptionHandler/
MavsThreadFactory/MavsThreadPoolExecutor/MavsThreadPoolStateMonitor源码

package com.landon.mavs.example.util;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
*
* Mavs Cache线程池 {@link java.util.concurrent.Executors#newCachedThreadPool()}
*
* @author landon
*
*/

publicclass MavsCachedThreadPoolExecutor extends MavsThreadPoolExecutor {

   
public MavsCachedThreadPoolExecutor() {
       
super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
               
new SynchronousQueue<Runnable>());
    }


   
public MavsCachedThreadPoolExecutor(ThreadFactory threadFactory) {
       
super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
               
new SynchronousQueue<Runnable>(), threadFactory);
    }


   
public MavsCachedThreadPoolExecutor(ThreadFactory threadFactory,
            RejectedExecutionHandler rejectHandler) {
       
super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
               
new SynchronousQueue<Runnable>(), threadFactory, rejectHandler);
    }

}



package com.landon.mavs.example.util;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
*
* 封装的固定线程数量的Mavs线程池
* {@link java.util.concurrent.Executors#newFixedThreadPool(int)}
*
* @author landon
*
*/

publicclass MavsFixedThreadPoolExecutor extends MavsThreadPoolExecutor {

   
public MavsFixedThreadPoolExecutor(int nThreads) {
       
super(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
               
new LinkedBlockingQueue<Runnable>());
    }


   
public MavsFixedThreadPoolExecutor(int nThreads, ThreadFactory threadFactory) {
       
super(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
               
new LinkedBlockingQueue<Runnable>(), threadFactory);
    }


   
public MavsFixedThreadPoolExecutor(int nThreads,
            ThreadFactory threadFactory, RejectedExecutionHandler rejectHandler) {
       
super(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
               
new LinkedBlockingQueue<Runnable>(), threadFactory,
                rejectHandler);
    }

}



package com.landon.mavs.example.util;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
* Mavs线程池拒绝执行策略
*
* @author landon
*
*/

publicclass MavsRejectedExecutionPolicy implements RejectedExecutionHandler {
   
privatestaticfinal Logger LOGGER = LoggerFactory
            .getLogger(MavsRejectedExecutionPolicy.class);

    @Override
   
publicvoid rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        LOGGER.debug("task rejectedExecution.ThreadPool.state:{}",
                MavsThreadPoolStateMonitor.monitor(executor));
    }

}



package com.landon.mavs.example.util;

import java.lang.Thread.UncaughtExceptionHandler;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
* Mavs线程默认的异常终止处理器
* {@link java.lang.ThreadGroup#uncaughtException(Thread, Throwable)}
* 中对于ThreadDeath的处理
*
* @author landon
*
*/

publicclass MavsThreadDefaultUncaughtExceptionHandler implements
        UncaughtExceptionHandler {
   
privatestaticfinal Logger LOGGER = LoggerFactory
            .getLogger(MavsThreadDefaultUncaughtExceptionHandler.class);

    @Override
   
publicvoid uncaughtException(Thread t, Throwable e) {
        LOGGER.warn("Exception in thread \"" + t.getName() + "\"", e);
    }


}



package com.landon.mavs.example.util;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
*
* Mavs线程工厂 {@link java.util.concurrent.Executors#defaultThreadFactory()} 参考
* {@link java.util.concurrent.Executors$DefaultThreadFactory}实现
*
* @author landon
*
*/

publicclass MavsThreadFactory implements ThreadFactory {
   
privatestaticfinal String MAVS_NAME_PREFIX ="Mavs-";

   
/** 线程号 */
   
privatefinal AtomicInteger threadNumber =new AtomicInteger(1);
   
/** 线程组 */
   
privatefinal ThreadGroup threadGroup;
   
/** 线程名字前缀 */
   
privatefinal String namePrefix;

   
/**
     *
     * 构造MavsThreadFactory
     *
     * @param processPrefix
     *            进程前缀
     * @param threadName
     *            线程名
    
*/

   
public MavsThreadFactory(String processPrefix, String threadName) {
        SecurityManager sm = System.getSecurityManager();
        threadGroup = (sm !=null) ? sm.getThreadGroup() : Thread
                .currentThread().getThreadGroup();

        namePrefix = MAVS_NAME_PREFIX + processPrefix +"-"+ threadName +"-";
    }


    @Override
   
public Thread newThread(Runnable r) {
        Thread t =new Thread(threadGroup, r, namePrefix
               
+ threadNumber.getAndIncrement(), 0);

       
// 做这两个设置的原因在于线程的daemon/priority属性默认是由Thread.currentThread决定

       
if (t.isDaemon()) {
            t.setDaemon(false);
        }


       
if (t.getPriority() != Thread.NORM_PRIORITY) {
            t.setPriority(Thread.NORM_PRIORITY);
        }


       
// 设置Mavs线程默认的异常终止处理器
        if (Thread.getDefaultUncaughtExceptionHandler() ==null) {
            Thread.setDefaultUncaughtExceptionHandler(new MavsThreadDefaultUncaughtExceptionHandler());
        }


       
return t;
    }


   
public String getNamePrefix() {
       
return namePrefix;
    }

}




package com.landon.mavs.example.util;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*
* Mavs线程池,提供了钩子方法的默认实现
*
* @author landon
*
*/

publicclass MavsThreadPoolExecutor extends ThreadPoolExecutor {
   
privatestaticfinal Logger LOGGER = LoggerFactory
            .getLogger(MavsThreadPoolExecutor.class);

   
public MavsThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
           
long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
       
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }


   
public MavsThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
           
long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
       
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                handler);
    }


   
public MavsThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
           
long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
       
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory);
    }


   
public MavsThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
           
long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
       
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory, handler);
    }


    @Override
   
protectedvoid beforeExecute(Thread t, Runnable r) {
        LOGGER.info("Thread["+ t.getName() +"]#beforeExecute:{}",
                MavsThreadPoolStateMonitor.monitor(this));
       
super.beforeExecute(t, r);
    }


    @Override
   
protectedvoid afterExecute(Runnable r, Throwable t) {
       
super.afterExecute(r, t);
        LOGGER.info("Thread["+ Thread.currentThread().getName()
               
+"]EfterExecute:{}", MavsThreadPoolStateMonitor.monitor(this));
       
if (t !=null) {
            LOGGER.warn("Worker.runs.task.err", t);
        }

    }


    @Override
   
protectedvoid terminated() {
       
super.terminated();
        LOGGER.info("terminated:{}", MavsThreadPoolStateMonitor.monitor(this));
    }

}



package com.landon.mavs.example.util;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
*
* Mavs线程池状态Monitor
*
* @author landon
*
*/

publicclass MavsThreadPoolStateMonitor {
   
/**
     *
     * 线程池状态监视
     *
     * @param executor
     * @return
    
*/

   
publicstatic String monitor(ThreadPoolExecutor executor) {
       
if (executor ==null) {
           
thrownew NullPointerException();
        }


       
// 核心线程数
        int corePoolSize = executor.getCorePoolSize();
       
// 最大线程数
        int maximumPoolSize = executor.getMaximumPoolSize();
       
// 线程保持活动时间
        long keepAliveTime = executor.getKeepAliveTime(TimeUnit.MILLISECONDS);

       
// 当前线程数
        int poolSize = executor.getPoolSize();
       
// 返回活跃(正在执行任务)的近似线程数
        int activeThreadCount = executor.getActiveCount();
       
// 返回曾经同时位于池中的最大线程数(包括已被回收的worker线程计数)
        int largestPoolSize = executor.getLargestPoolSize();

       
// 已完成执行的近似任务总数
        long completedTaskCount = executor.getCompletedTaskCount();
       
// 曾计划完成的近似任务总数(completedTaskCount + 工作队列大小 + 正在执行任务的worker线程数目)
        long taskCount = executor.getTaskCount();
       
// 工作队列大小
        int workQueueSize = executor.getQueue().size();

       
// 是否在非RUNNING状态下
        boolean isShutdown = executor.isShutdown();
       
// 是否是TERMINATED状态
        boolean isTerminated = executor.isTerminated();
       
// 是否是SHUTDOWN或者STOP状态
        boolean isTerminating = executor.isTerminating();

        String executorName
="Default-ThreadPoolExecutor";
        ThreadFactory factory
= executor.getThreadFactory();
       
if (factory !=null&& factory instanceof MavsThreadFactory) {
            executorName
= ((MavsThreadFactory) factory).getNamePrefix();
        }


       
return executorName +" [corePoolSize="+ corePoolSize
               
+", maximumPoolSize="+ maximumPoolSize +", keepAliveTime="
               
+ keepAliveTime +", poolSize="+ poolSize
               
+", activeThreadCount="+ activeThreadCount
               
+", largestPoolSize="+ largestPoolSize
               
+", completedTaskCount="+ completedTaskCount +", taskCount="
               
+ taskCount +", workQueueSize="+ workQueueSize
               
+", isShutdown="+ isShutdown +", isTerminated="
               
+ isTerminated +", isTerminating="+ isTerminating +"]";
    }

}



三.总结:
    本篇结合jdk源码重点讲述了ThreadPoolExecutor和ExecutorService.另外提供了本人封装的一些线程池相关的库源码.

分享到:
评论

相关推荐

    Java多线程详解(超详细)_狂神说笔记完整版_项目代码_适合小白随课程学习

    Java多线程详解 在Java编程中,多线程是一种重要的技术,它使得程序能够同时执行多个任务,提高系统的效率和响应性。本教程将详细讲解Java中的多线程概念,包括线程的创建、状态、同步以及高级主题,旨在帮助初学者...

    JAVA 多线程学习笔记

    这篇学习笔记将深入探讨Java多线程的核心概念、实现方式以及相关工具的使用。 一、多线程基础 1. 线程与进程:在操作系统中,进程是资源分配的基本单位,而线程是程序执行的基本单位。每个进程至少有一个主线程,...

    多线程笔记_java/多线程_

    在Java编程语言中,多线程是程序设计中的一个重要概念,尤其在开发高效能和响应迅速的应用时。本文档将全面解析多线程的基础知识,从简单到复杂,帮助开发者深入理解并掌握这一核心技术。 一、多线程基础 1.1 线程...

    Java多线程自学笔记

    ### Java多线程自学笔记知识点总结 #### 一、Java多线程概念 - **多线程定义**:多线程是指程序中可以同时执行多个线程的能力。Java虚拟机支持多线程机制,允许应用程序在不同的线程上并行执行任务。 - **线程特点...

    多线程精品资源--java-study 是本人学习Java过程中记录的一些代码!从Java基础的数据类型、jdk1..zip

    ”则提示这是一份个人学习笔记,包含了作者在掌握Java多线程编程时编写的示例代码。 【描述】中的信息虽然不完整,但从“从Java基础的数据类型、jdk1.”我们可以推测,这个资源不仅限于多线程部分,可能还涵盖了...

    java多线程编程笔记

    学习Java多线程编程不仅涉及到线程的创建和启动,还包括线程同步(如`synchronized`关键字、`wait()`, `notify()`, `notifyAll()`方法)、线程间通信(`BlockingQueue`、`Future`、`ExecutorService`)、线程池(`...

    Java多线程源码笔记.pdf

    Java多线程是Java编程中的重要概念,它允许程序同时执行多个任务,从而提高系统效率。在Java中,实现多线程主要有两种方式:通过继承Thread类和实现Runnable接口。Thread类提供了创建新线程的基本功能,而Runnable...

    java多线程相关资料整理

    Java多线程是Java开发中的重要组成部分,它允许程序同时执行多个任务,提高..."Java多线程源码笔记.pdf"这样的文档通常会涵盖以上内容,并可能包含具体的源码分析和实战案例,对于理解和掌握Java多线程编程非常有帮助。

    多线程-狂神说Java学习笔记

    本学习笔记将深入探讨Java多线程的相关知识,包括其原理、实现方式、同步机制以及常见问题。 ### 一、多线程的基本概念 多线程是指在一个程序中存在两个或更多的执行线程,这些线程共享同一内存空间,但各自拥有...

    java线程学习笔记

    Java线程学习笔记涉及了Java多线程编程的多个关键知识点,本篇知识点整理将详细解释每个概念及其在Java中的实现方式。 基本知识部分包含了Java线程编程的基础内容,它们是并发编程的基石。 任务Runnable是一个接口...

    java从入门到精通学习笔记-Lesson5(多线程).pdf

    Java 多线程是编程中的重要概念,尤其对于并发处理和高效系统设计至关重要。在Java中,多线程是通过两种主要方式实现的:继承`Thread`类或实现`Runnable`接口。 首先,理解基本概念:程序是静态的指令集合,而进程...

    JUC学习笔记(Java多线程)

    本笔记将深入探讨JUC的相关知识点,帮助开发者理解并掌握Java多线程的核心概念。 首先,我们来了解一下Java线程。线程是程序执行的基本单元,允许多个任务在同一时间运行,从而提高了系统资源的利用率。在Java中,...

    JAVA多线程共13页.pdf.zip

    这个"JAVA多线程共13页.pdf.zip"压缩包文件很可能包含了一份关于Java多线程的详细教程或笔记,虽然具体内容无法直接展示,但我们可以深入探讨Java多线程的相关知识点。 1. **线程的概念**:线程是操作系统分配CPU...

    马士兵多线程训练营笔记

    6. **线程池**:ExecutorService、ThreadPoolExecutor和Future接口的理解,线程池的配置参数及其优化策略。 7. **并发工具类**:CountDownLatch、CyclicBarrier、Semaphore、Exchanger等并发工具的用法及应用场景。...

    java私塾全笔记+1-15章pdf

    6. **多线程**: - **线程的创建与运行**:通过Thread类和Runnable接口创建线程,理解start()与run()的区别。 - **线程同步**:synchronized关键字的使用,wait()、notify()和notifyAll()的调用,以及死锁问题的...

    Java随堂笔记

    6. **多线程** - **线程的创建**:通过实现Runnable接口或继承Thread类。 - **线程同步**:synchronized关键字、wait()、notify()和notifyAll()方法,以及Lock接口。 - **线程池**:ExecutorService、...

    java笔记 java笔记 java笔记

    5. **多线程编程**: - **线程创建**:通过继承Thread类或实现Runnable接口创建线程。 - **线程同步**:synchronized关键字、wait/notify机制以及Lock接口。 - **线程池**:ExecutorService接口和...

    张龙JAVA圣思园笔记

    Java对多线程的支持是其强大特性之一。笔记中可能会介绍Thread类的创建与管理,同步控制(synchronized关键字、wait/notify机制、Lock接口)、线程池(ExecutorService、ThreadPoolExecutor)等知识点。 3. **序列...

    达内CORE JAVA部分笔记,代码,练习

    6. **多线程** - Java提供了Thread类和Runnable接口来实现多线程,理解线程同步和互斥是并发编程的基础。 - 线程池:ExecutorService和ThreadPoolExecutor等类提供了线程管理的高级功能。 7. **反射机制** - ...

    Java学习超强笔记

    7. **多线程** - 线程的创建:通过Thread类或实现Runnable接口。 - 线程同步:synchronized关键字,wait()、notify()和notifyAll()方法,以及Lock接口。 - 线程池:ExecutorService,ThreadPoolExecutor和...

Global site tag (gtag.js) - Google Analytics