JUC 线程池

2019-02-13 本文已影响43人黄靠谱

概述

线程池的作用：节省资源、提升响应、削峰限流、管理线程
ThreadPoolExecutor的核心参数：corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、handler
Executors静态工厂类，提供常用的4种线程池：

FixedThreadPool,固定线程数量的线程池，配合一个无界的LinkedBlockingQueue
SingleThreadPool,固定使用一个线程来工作，但是可以无限堆积，因为是单个线程，所以它可以保证认为是按顺序执行的，因为LinkedBlockingQueue是严格的FIFO的
CachedThreadPool，同步线程池,默认非公平的,基于SynchronousQueue队列
ScheduledThreadPool,指定核心线程数的定时执行线程池，使用时通过 pool.schedule调用，指定延迟时间，依赖DelayedWorkQueue来实现

线程池使用的优先级：基本线程池 > BlockingQueue >最大线程池 >饱和策略
线程池支持的函数

execute 不会返回线程执行是否成功
submit 返回一个future对象可以阻塞获取返回值
shutdown的原理是只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。

类的继承关系

http://cmsblogs.com/?p=2444

Executor作为顶层接口，只提供一个execute()接口方法
ThreadPoolExecutor继承AbstractExecutorService抽象类-->ExcutorService接口-->Executor顶层接口
Executors静态工厂类，提供了很多的通用线程池的静态方法

线程池的作用

节省资源：复用线程，减少重复创建和销毁线程
提升响应速度：如果有可以复用的线程，不需要创建新线程，直接执行业务操作，提升速度
方便管理，提供了线程池的shutdown、shutdownnow、isTerminated、isShutDown等方法
削峰限流，不让一下子过多的线程把程序搞崩，比如RocketMQ里面的Consumer就会把获取到的消息放入到线程池里来慢慢消费,Dubbo也有一个默认100长度的LinkedBlockQueue

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

使用规则

优先级：基本线程池 > BlockingQueue >最大线程池 >饱和策略

线程池判断基本线程池是否已满？没满，创建一个工作线程来执行任务，即使有空闲的已经创建的线程也不会复用。满了，则进入下个流程。
线程池判断工作队列是否已满？没满，则将新提交的任务存储在工作队列里。满了，则进入下个流程
线程池判断整个线程池是否已满？没满，则创建一个新的工作线程来执行任务，满了，则交给饱和策略来处理这个任务

runnableTaskQueue

常用阻塞队列特性总结：通过加锁实现安全地读写（size、contains函数也是加锁读），通过condition或者AtomicInteger来协调生产和消费

ArrayBlockQueue：1把锁，无法并发写，2个condition，通过await、signal来实现线程调度。FIFO队列的定长阻塞队列
LinkedBlockQueue：2把锁，支持生产、消费并发执行，通过AtomicInteger和CAS来控制库存。可以指定大小，默认无界队列。
SynchronousQueue：没有容量的队列，put、take成为一对儿才可以执行，否则阻塞，add和remove如果没有配对成功，直接报错。用于快速响应的业务场景
PriorityBlockingQueue：有优先级的队列，插入元素实现Compare接口，内部维护了一个最小堆，一把锁，基于数组，自动扩容的无界队列

RejectedExecutionHandler(饱和策略)

AbortPolicy，默认饱和策略，直接抛出异常
CallerRunsPolicy：只用调用者所在线程来运行任务
DiscardPolicy：不处理，丢弃掉，也不会日志什么的
DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
自定义Police，需要实现RejectedExecutionHandler

线程池的使用 execute submit

execute 不会返回线程执行是否成功

threadsPool.execute(new Runnable() {}）；

submit 返回一个future对象可以阻塞获取返回值

Future future=threadsPool.submit(new Runnable() {}）；
Boolean result= future.get();

线程池的关闭

shutdown或shutdownNow方法来关闭线程池，但是它们的实现原理不同，

shutdown的原理是只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。
shutdownNow的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。shutdownNow会首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表。

Callable &Future &FutureTask

http://blog.csdn.net/javazejian/article/details/50896505

Callable接口

加强版的Runnable,有返回值，会抛出异常

public interface Runnable {  
    public abstract void run();  
}  
public interface Callable<V> {   
      V   call()   throws Exception;   
}

可以作为参数，被线程池调用执行，Future<T> submit(Callable<T> task);

Future接口

作为异步计算的顶层接口，Future对具体的Runnable或者Callable任务提供了三种操作：

执行任务的取消：cancel()
查询任务的状态: isDone()、isCancelled()
获取任务的执行结果（只有Callable任务才有，而且是阻塞）：get()、V get(Long timeout , TimeUnit unit)

FutureTask类(同时实现了Runnable、Future)

既可以作为一个Runnable实现类在线程池中执行，甚至可以直接new Thread(futureTask).start();也可以作为一个future获取其他线程的执行结果和状态，
done():异步计算完成后调用的回调函数

数据库连接池原理

解决的问题:频繁的建立、关闭连接，会极大的减低系统的性能，因为对于连接的使用成了系统性能的瓶颈。解决方案：连接复用。
大部分连接池可以设置最大连接数，最小连接数，最大闲置时间等参数，连接池根据这些参数，新建和关闭连接

使用案例

Executors 类里面实现了一些常用的工具方法，返回适合不同业务场景的线程池实现类。
但是要注意FixedThreadPool和SingleThreadPool都是定长的线程池，但是排队队列都是无限长的；CachedThreadPool和ScheduledThreadPool都是可以启动无数个线程来完成功能，都要注意不能OOM了。

FixedThreadPool,固定线程数量的线程池，配合一个无界的LinkedBlockingQueue，重复复用指定数量的coresize的线程

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

SingleThreadPool,固定使用一个线程来工作，但是可以无限堆积，因为是单个线程，所以它可以保证认为是按顺序执行的，因为LinkedBlockingQueue是严格的FIFO的

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

CachedThreadPool，同步线程池,默认非公平的

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

ScheduledThreadPool,指定核心线程数的定时执行线程池，使用时通过 pool.schedule调用，指定延迟时间，依赖DelayedWorkQueue来实现

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

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



ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);

        for (int i = 0; i < 5; i++) {
            Future<Integer> result = pool.schedule(new Callable<Integer>() {

                @Override
                public Integer call() throws Exception {
                    int num = new Random().nextInt(100);//生成随机数
                    System.out.println(Thread.currentThread().getName() + " : " + num);
                    return num;
                }
            }, 3, TimeUnit.SECONDS);


            System.out.println(result.get());
        }

        pool.shutdown();

数据库的连接池管理

https://www.cnblogs.com/happySmily/p/5941813.html

常见的数据库连接池：dbcp、c3p0、druid | weblogic（收费的）
常见参数：

initialSize=10：程序一启动就自动创建 N个数据库长连接
timeBetweenEvictionRunsMillis = "30000" ，程序启动后，每隔30scheck一下当前线程数是否合格:A）Idle线程是否超时，超时就回收掉 B)当前可用线程是否大于minIdle，小于就自动创建，如果设置为非正数,则不运行空闲连接回收器线程，默认是-1，不启动单独线程维护minIdle
maxIdle：当可用线程>minIdle但是小于maxIdle的时候，空闲过期就会被回收
maxActive：当可用线程超过maxIdle的，小于maxActive的时候，使用完就立刻回收
maxWait=3000；这个指的是最大等待时间，如果超过3秒还未分配到连接，就报错

maxWait="3000"    从池中取连接的最大等待时间，单位ms.  
initialSize="10"  初始化连接    
minIdle="10"   最小空闲连接  
maxIdle="60"   最大空闲连接
maxActive="80" 最大活动连接  

validationQuery = "SELECT 1"  验证使用的SQL语句  
testWhileIdle = "true"      指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除.  
testOnBorrow = "false"   借出连接时不要测试，否则很影响性能  
timeBetweenEvictionRunsMillis = "30000"  每30秒运行一次空闲连接回收器  
minEvictableIdleTimeMillis = "1800000"  池中的连接空闲30分钟后被回收  
numTestsPerEvictionRun="10" 在每次空闲连接回收器线程(如果有)运行时检查的连接数量  
      
removeAbandoned="true"  连接泄漏回收参数，当可用连接数少于3个时才执行  
removeAbandonedTimeout="180"  连接泄漏回收参数，180秒，泄露的连接可以被删除的超时值

参考资料

大牛博客：宏观上分析线程池
http://ifeve.com/java-threadpool/

CSDN系列专栏简易
http://blog.csdn.net/column/details/javathreadpool.html

死磕java并发系列有4篇线程池相关的
http://cmsblogs.com/?p=2122