支持生产阻塞的线程池, 使用了阻塞生产者的方式. 把队列设为有限队列.队列满了,调用构造时传入的RejectedExecutionHandler去拒绝任务的处理
RejectedExecutionHandler中继续往有界队列中put(阻塞)来添加元素.
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (!executor.isShutdown()) {
try {
executor.getQueue().put(r);
} catch (InterruptedException e) {
// should not be interrupted
}
}
}
};
http://ifeve.com/blocking-threadpool-executor/
复习一下线程池队列满了之后的拒绝策略
一复习,拉一串.
谷歌的有个库,
提供了一个ThreadFactoryBuilder,可以把一些复杂的参数都写在链式里面.
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("ABC-Pool-%d").build();
然后创建线程池
ExecutorService pool = new ThreadPoolExecutor(
3, //3个core
200,//200个maxPoolSize
0L,//keepAlived time
TimeUnit.MILLISECONDS, //kepAlivedTime 单位
new LinkedBlockingQueue<Runnable>(1024), //队列, 设为有界
namedThreadFactory, //线程工厂
new ThreadPoolExecutor.AbortPolicy() //拒绝策略, 是RejectedExecutionHandler 的实现
);
其中JDK实现的拒绝策略有4种
AbortPolicy 的实现其实是抛出 throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());
DiscardPolicy 的实现是 什么都不做,-->悄悄的丢掉
DiscardOldestPolicy 的实现是, 悄悄的从队列里e.getQueue().poll();掉一个task, 然后把当前task加进去. --->悄悄的顶一个出去.
CallerRunsPolicy的实现是, 让生产者自己做.(生产线程就会阻塞)
线程池然后使用 pool.execute(runnable)来执行runnable task.
pool.execute(()->{ //do something });
可以结合countDownLatch,来阻塞最后的终止.
-----------------------------------------------------------------------------------------
interface ExecutorService extends Executor
executor.execute(()->{});//是执行runnable
executorService.submit 可以submit callabel和runnable
--------------------------------------------------------------------------------------------------------
CompletableFuture.runAsync参数是一个runnable,
内部实现是:
--->
asyncRunStage(asyncPool, runnable);
--->
asyncRunStage(screenExecutor(executor), runnable);
--->
CompletableFuture<Void> d = new CompletableFuture<Void>();
executor.execute(new AsyncRun(d, f));
AsyncRun 是一个静态内部类,implements Runnable和Task系某接口
static final class AsyncRun extends ForkJoinTask<Void> implements Runnable, AsynchronousCompletionTask {
executor 是一个线程池
private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
screenExecutor(executor) 其实就是null-check
Null-checks user executor argument, and translates uses of commonPool to asyncPool in case parallelism disabled.