首先，阿里的Java规范不推荐使用类Executors的静态方法创建线程池。至于为什么不推荐，通过看源码，如下：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
可以得知原因如下：
-
newFixedThreadPool, newSingleThreadExecutor两种方法没有限制LinkedBlockingQueue队列的大小，当任务越来越多时，队列的大小会越来越大，可能会造成OOM。 -
newCachedThreadPool, newScheduledThreadPool两种方法的线程最大数量为Integer.MAX_VALUE，当任务越来越多时，也可能会造成OOM。
以上四种静态方法实际上都是通过new ThreadPoolExecutor()实现的，所以我们可以根据实际需要创建符合自己要求的线程池，其中一个构造方法如下：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
...
}
下面解释一下构造参数解释：
-
corePoolSize：核心池大小 -
maximumPoolSize：线程池最大值 -
keepAliveTime：非核心池的线程执行完毕后存活时间，可以通过pool.allowCoreThreadTimeOut(true)方法设置核心池中线程也适用于存活时间限制。 -
unit：存活时间的单位 -
workQueue：任务队列，队列有很多种，比如ArrayBlockingQueue、LinkedBlockingDeque、SynchronousQueue等三种队列，前面两种就是数据结构不同，前者数组，后者链表，主要是第三个同步队列有点特别。（SynchronousQueue是一种阻塞队列，每个插入操作必须等待另一个线程的移除操作，反之亦然。任务不会在队列中停留，所以其内部没有容量，size()方法永远为0。） -
threadFactory：线程工厂，创建线程的工厂类 -
handler：拒绝策略，拒绝策略有四种：-
AbortPolicy: 丢弃新任务，并抛出异常，默认策略 -
DiscardPolicy: 丢弃新任务，不抛出异常 -
CallerRunsPolicy: 交由调用execute方法的线程去执行新任务 -
DiscardOldestPolicy: 接收新任务，但会丢弃队列头部的任务
-
介绍具体参数之后，先看下线程池执行任务的流程：
- 先创建线程去执行任务，当
线程数量=corePoolSize，后续的新任务都丢进队列， - 当队列也满了，如果
maximumPoolSize>corePoolSize，继续创建线程执行新任务，否则执行拒绝策略。 - 当
线程数量=maximumPoolSize，后续的任务执行拒绝策略。
线程池流程图.png
再用例子验证上面所述流程，下面是一个默认拒绝策略的例子：
public class ThreadPoolTest {
public static void main(String[] args) {
LinkedBlockingDeque<Runnable> deque = new LinkedBlockingDeque<>(3);
ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 4, 0, TimeUnit.MILLISECONDS, deque);
for (int i = 0; i < 8; i++) {
System.out.println("PoolSize: "+ pool.getLargestPoolSize());
System.out.println("dequeSize: "+ deque.size());
int b = i;
pool.submit(() -> {
System.out.println(b+"is run...");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
});
}
pool.shutdown();
}
}
打印结果：
PoolSize: 0
dequeSize: 0
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@7ba4f24f rejected from java.util.concurrent.ThreadPoolExecutor@3b9a45b3[Running, pool size = 4, active threads = 4, queued tasks = 3, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
at cn.tl.thread.ThreadPoolTest.main(ThreadPoolTest.java:29)
PoolSize: 1
dequeSize: 0
PoolSize: 2
0is run...
1is run...
dequeSize: 0
PoolSize: 3
dequeSize: 0
2is run...
PoolSize: 3
dequeSize: 1
PoolSize: 3
dequeSize: 2
PoolSize: 3
dequeSize: 3
PoolSize: 4
dequeSize: 3
6is run...
3is run...
5is run...
4is run...
很明显，当执行第8次循环时，线程数量=maximumPoolSize，线程池执行了默认拒绝策略AbortPolicy: 丢弃新任务，并抛出异常。所以设置线程数和队列容量的时候需根据实际需求指定。
再看一个同步队列和CallerRunsPolicy策略的示例：
public class SynchronousQueueTest {
public static void main(String[] args) {
ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());
for (int i = 0; i < 5; i++) {
int b = i;
pool.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " " + b);
});
}
pool.shutdown();
}
}
打印结果：
main 2
pool-1-thread-1 0
pool-1-thread-2 1
main 3
pool-1-thread-2 4
该示例中，线程池只能同时执行2个任务，多余的任务暂时交给主线程执行，当线程池有空闲的线程，就继续执行新任务。
线程池默认命名不适合查看日志，可以根据需求设置自己的线程名，引入第三方的jar包，如下：
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("demo-pool-%d").build();
ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 4, 0, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(3), namedThreadFactory);
<!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
以上。
网友评论