浅谈java线程池
2017-02-27 本文已影响117人
扈扈哈嘿
线程池有啥好处
- 降低资源消耗：通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提高响应速度：当任务到达时，任务可以不需要等到线程创建就能立即执行。
- 提高线程的可管理性：可以有效的控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。提供定时执行，定期执行，单线程，并发控制等功能。
介绍一下ThreadPoolExecutor
构造器与参数
在java源码中它有四个构造器，分别如下：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
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;
}
最终都是调用最下面那个构造方法，只是参数不同罢了。那现在来说说这些参数都是什么意思。
- corePoolSize:线程池中所保存的核心线程数，包括空闲线程。
- maximumPoolSize：线程池允许创建的最大线程数。
- keepAliveTime：线程池的工作线程空闲时间。线程池的工作线程空闲后，保持存活的时间。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。
- TimeUnit:线程活动保持时间的单位。可选的单位有天，小时，分钟等。
-
BlockingQueue:任务执行前保存任务的队列，仅保存由execute方法提交的Runnale任务。
队列的种类：
- ArrayBlockingQueue: 是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。
- LinkedBlockingQueue:一个基于链表结构的阻塞队列，此队列按FIFO原则对元素进行排序，吞吐量通常高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
- SynchronousQueue:一个不存储元素的阻塞队列。每插入操作必须等到另一下线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue,静态工厂方法Executors.newCachedThreadPool使用了这个队列。
- PriorityBlockingQueue:一个具有优先级的无限阻塞队列。
它与corePoolSize和maximumPoolSize的关系如下：
线程池处理流程.jpg
- ThreadFactory ：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字和优先级等。
- RejectedExecutionHandler ：饱和策略。当线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy,表示无法处理新任务时抛出异常。以下是JDK1.5提供的四种策略。
- AbortPolicy：直接抛出异常。
- CallerRunsPolicy:只用调用者所在线程来运行任务
- DiscardOldestPolicy 丢弃队列里最近的一个任务，并执行当前任务。
- DiscardPolicy 不处理，丢弃掉
当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略，如记录日志或持久化不能处理的任务。
2 执行方法
- execute方法
我们可以使用execute提交任务，但是execute没有返回值，所以无法判断任务是否被线程池执行成功。
threadsPool.execute(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
});
- submit 方法
我们也可以使用submit方法来提交任务，它会返回一个future，那么我们可以通过这个future来判断任务是否执行成功，通过future的get方法来获取返回值，get方法会阻塞住直到任务完成，而使用get(long timeout, TimeUnit unit)方法则会阻塞一段时间后立即返回，这时有可能任务没有执行完。
Future<Object> future = executor.submit(task);
try {
Object s = future.get();
} catch (InterruptedException e) {
// 处理中断异常
} catch (ExecutionException e) {
// 处理无法执行任务异常
} finally {
// 关闭线程池
executor.shutdown();
}
- shutdown与shutdownNow
通过调用shutdown或shutdownNow方法来关闭线程，它们的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。但它们存在一定的区别，shutdownNow首先将线程池的状态设置成stop,，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表，而shutdown只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。
Executors与ThreadPoolExecutor
Executors提供四种线程池
- newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。
- **newFixedThreadPool **
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。定长线程池的大小最好根据系统资源进行设置，如Runtime.getRuntime().availableProcessors()。
- **newScheduledThreadPool **
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
ScheduledThreadPoolExecutor是ThreadPoolExecutor的子类。
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
其中的队列是DelayedWorkQueue,支持定时及周期性任务执行.
比如说：
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
// 表示延迟3秒执行
scheduledThreadPool.schedule(new Runnable() {
public void run() {
System.out.println("delay 3 seconds");
}
}, 3, TimeUnit.SECONDS);
}
// 表示延迟1秒后每3秒执行一次
scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
public void run() {
System.out.println("delay 1 seconds, and excute every 3 seconds");
}
}, 1, 3, TimeUnit.SECONDS);
}
- **newSingleThreadExecutor **
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有的任务按指定顺序执行。
查看Executors源码我们知道，Executors类提供了使用ThreadPoolExecutor 的简单的 ExecutorService 实现，也就是上面所说的四种Executors线程池，但是 ThreadPoolExecutor 提供的功能远不止于此。 不过在Java doc中，并不提倡我们直接使用ThreadPoolExecutor，而是使用Executors类中提供的几个静态方法来创建线程池。
我们可以在创建 ThreadPoolExecutor 实例时指定活动线程的数量，我们也可以限制线程池的大小并且创建我们自己的 RejectedExecutionHandler 实现来处理不能适应工作队列的工作。
合理配置线程池大小
遵循两原则：
1、如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU+1
2、如果是IO密集型任务，参考值可以设置为2*NCPU
当然，这只是一个参考值，具体的设置还需要根据实际情况进行调整，比如可以先将线程池大小设置为参考值，再观察任务运行情况和系统负载、资源利用率来进行适当调整。
