本文介绍java中的线程池类ThreadPoolExecutor。
我们可以利用ThreadPoolExecutor创建线程池,这个类中有多个构造方法。
-
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)
-
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)
-
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)
-
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
参数介绍:
corePoolSize : 核心池的大小,即核心线程的数目
maximumPoolSize:线程池最大线程数
keepAliveTime:空闲线程存活时间。表示线程没有任务执行时,多久会终止。
unit:对应keepAliveTime的时间单位
workQueue:阻塞队列,存储等待执行的任务
threadFactory:创建线程时所使用的工厂
handler:拒绝处理任务时的策略,有以下四种取值:
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常
ThreadPoolExecutor.CallerRunsPolicy:让调用者运行任务
ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列中最早的任务,本任务取而代之
ThreadPoolExecutor.DiscardPolicy:丢弃任务,不抛异常
当线程池的线程数大于corePoolSize时,就会使用之前的空闲线程。空闲线程数 = maximumPoolSize - corePoolSize,空闲线程有存活时间keepAliveTime,核心线程则没有。
线程提交方法:execute()
线程池使用例子:
public class Test1 {
private static final Logger log = LoggerFactory.getLogger(Test1.class);
public static void main(String[] args) throws InterruptedException {
// 自定义线程创建工厂
MyThreadFactory myThreadFactory = new MyThreadFactory();
// 新建阻塞队列,容量为1
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(1);
// 使用DiscardPolicy策略,拒绝任务时直接丢弃
RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
// 创建线程池,核心池2,总3,空闲1,存活时间5s
ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 3, 5, TimeUnit.SECONDS, queue, myThreadFactory, handler);
executor.execute(() -> {
try {
Thread.sleep(1000);
log.debug("1");
} catch (InterruptedException e) {
log.debug("被打断1");
}
});
executor.execute(() -> {
try {
Thread.sleep(1000);
log.debug("2");
} catch (InterruptedException e) {
log.debug("被打断2");
}
});
executor.execute(() -> {
try {
Thread.sleep(1000);
log.debug("3");
} catch (InterruptedException e) {
log.debug("被打断3");
}
});
executor.execute(() -> {
try {
Thread.sleep(1000);
log.debug("4");
} catch (InterruptedException e) {
log.debug("被打断4");
}
});
executor.execute(() -> {
try {
Thread.sleep(1000);
log.debug("5");
} catch (InterruptedException e) {
log.debug("被打断5");
}
});
}
// 线程创建工厂
static class MyThreadFactory implements ThreadFactory {
private AtomicInteger atomic = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "myThread-" + atomic.getAndIncrement());
}
}
}
运行结果:
2021-05-04 17:34:08:323 [myThread-1] - 1
2021-05-04 17:34:08:323 [myThread-3] - 4
2021-05-04 17:34:08:323 [myThread-2] - 2
2021-05-04 17:34:09:348 [myThread-1] - 3
创建了三个线程输出了1 ,4,2,在log输出3的时候,放入了阻塞队列,阻塞队列已满,所以输出5的任务根据策略直接丢弃。
程序的运行没有停止,是因为线程池没有关闭。
根据业务需要,如果需要关闭线程池,可以使用以下方法:
- shutdown() :不会接收新任务,已提交任务会执行完,会打断不在执行中的任务
- shutdownNow(): 不会接收新任务,会将队列中的任务返回,并用interrupt打断所有任务。
将上面的例子1的main方法末尾,加上代码:
Thread.sleep(500);
executor.shutdown();
运行结果:
2021-05-04 17:34:48:154 [myThread-1] - 1
2021-05-04 17:34:48:154 [myThread-2] - 2
2021-05-04 17:34:48:169 [myThread-3] - 4
2021-05-04 17:34:49:162 [myThread-1] - 3
程序正确退出,等待队列中的任务也正常执行了。
将上面的例子1的main方法末尾,加上代码
Thread.sleep(500);
executor.shutdownNow();
运行结果:
2021-05-04 17:35:08:977 [myThread-1] - 被打断1
2021-05-04 17:35:08:977 [myThread-2] - 被打断2
2021-05-04 17:35:08:977 [myThread-3] - 被打断4
程序被打断,且阻塞队列中的任务也没有继续执行。
Executors
我们也可以利用Executors工具类来创建线程池,但是并不推荐这样使用。
-
newFixedThreadPool(int nThreads) :能够创建固定大小的线程池,每次有任务进来就会创建一个线程,直到达到线程池的最大值。阻塞队列LinkedBlockingQueue是无界的,可以放任意数量的任务。
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
-
newCachedThreadPool :核心线程是0,使用的都是空闲线程,每个线程都有存活时间。可以无限创建线程。阻塞队列使用的是SynchronousQueue,是一个缓冲区为1的阻塞队列。
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
-
newSingleThreadExecutor:线程数量为1的线程池,类似于newFixedThreadPool(1)。不同的是它的返回值又包装了一层FinalizableDelegatedExecutorService,对外只暴露了ExecutorService接口,不能调用ThreadPoolExecutor中特有的方法,而newFixedThreadPool对外暴露的是ThreadPoolExecutor对象,可以强转后调用相应的方法修改核心线程数等属性
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
ScheduledThreadPoolExecutor
当我们需要延时,定时执行线程的时候,就可以使用ScheduledThreadPoolExecutor来创建线程池。
ScheduledThreadPoolExecutor继承ThreadPoolExecutor ,实现了ScheduledExecutorService。
常用方法:
- 在延迟delay时间后执行任务command:
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
例子: 在三秒后执行输出的任务:
public static void main(String[] args) throws InterruptedException {
log.debug("main");
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
executor.schedule(()->{
log.debug("running...");
},3,TimeUnit.SECONDS);
executor.shutdown();
}
运行结果:
2021-05-07 22:12:51.001 [main] - main
2021-05-07 22:12:54.047 [pool-1-thread-1] - running...
- 周期性地执行任务:
initialDelay: 在给定的初始延迟后,开始执行任务
period:每次执行的周期时间
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
例子: 在0秒延迟后,每隔1秒输出
public static void main(String[] args) throws InterruptedException {
log.debug("main");
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
executor.scheduleAtFixedRate(() -> {
log.debug("running...");
}, 0, 1, TimeUnit.SECONDS);
}
运行结果:
2021-05-07 22:19:10.873 [main] - main
2021-05-07 22:19:10.909 [pool-1-thread-1] - running...
2021-05-07 22:19:11.911 [pool-1-thread-1] - running...
2021-05-07 22:19:12.911 [pool-1-thread-1] - running...
......(略)
每隔一秒输出,但是如果执行的任务超过了定时时间1S,会怎样? 把任务执行时间修改为执行三秒,在运行:
修改:
executor.scheduleAtFixedRate(() -> {
try {
Thread.sleep(3000);
log.debug("running...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 0, 1, TimeUnit.SECONDS);
运行结果:
2021-05-07 22:22:35.324 [main] - main
2021-05-07 22:22:38.362 [pool-1-thread-1] - running...
2021-05-07 22:22:41.364 [pool-1-thread-1] - running...
2021-05-07 22:22:44.364 [pool-1-thread-1] - running...
......(略)
从运行结果可以看出,程序每隔3秒执行。
程序执行时间大于定时时间的话, 上次任务执行完便会立马执行下一次任务。
如果业务需要上一次程序执行完之后,再开始计时的话,就可以使用scheduleWithFixedDelay方法:
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
例子:
public static void main(String[] args) throws InterruptedException {
log.debug("main");
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
executor.scheduleWithFixedDelay(() -> {
try {
Thread.sleep(3000);
log.debug("running...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 0, 1, TimeUnit.SECONDS);
}
运行结果:
2021-05-07 22:26:15.436 [main] - main
2021-05-07 22:26:18.479 [pool-1-thread-1] - running...
2021-05-07 22:26:22.481 [pool-1-thread-1] - running...
2021-05-07 22:26:26.483 [pool-1-thread-1] - running...
2021-05-07 22:26:30.484 [pool-1-thread-1] - running...
......(略)
从运行结果可以看出,程序每隔4s (执行时间3s + 定时时间1s)执行。
从上次执行结束后,再延迟delay时间后,才会开始执行下一次任务。
标签:10,log,22,05,编程,线程,2021,ThreadPoolExecutor From: https://www.cnblogs.com/Aeons/p/18375584