简介
线程池循环执行一些任务,某个线程执行超时,需要将超时的线程任务抛弃。
示例
修改前
当遇到超时的任务就凉凉,得重启程序。
Task.java:
public class Task implements Runnable {
private final int sleepTime;
private final CountDownLatch countDownLatch;
public Task(int sleepTime, CountDownLatch countDownLatch) {
this.sleepTime = sleepTime;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
Thread.sleep(sleepTime == Demo.THREAD_SIZE ? Demo.LONG_RUNNING_THREAD_TIME : Demo.SHORT_RUNNING_THREAD_TIME);
System.out.println("任务 " + sleepTime + " 干完了");
countDownLatch.countDown();
}
}
Demo.java:
public class Demo {
/**
* 线程数量
* 前三个任务执行 {@link Demo#SHORT_RUNNING_THREAD_TIME} ms
* 最后一个线程运行 {@link Demo#LONG_RUNNING_THREAD_TIME} ms
*/
public static final int THREAD_SIZE = 4;
/**
* 线程超时时间(ms)
*/
public static final int THREAD_TIMEOUT = 3000;
/**
* 超时线程运行的时间(ms)
*/
public static final int LONG_RUNNING_THREAD_TIME = 20000;
/**
* 正常线程运行的时间(ms)
*/
public static final int SHORT_RUNNING_THREAD_TIME = 10;
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newFixedThreadPool(1);
while (true) {
CountDownLatch countDownLatch = new CountDownLatch(THREAD_SIZE);
System.out.println("开始");
for (int i = 1; i <= THREAD_SIZE; i++) {
executorService.execute(new Task(i, countDownLatch));
}
if (!countDownLatch.await(Demo.THREAD_TIMEOUT, TimeUnit.MILLISECONDS)) {
throws new Exception("凉凉,重启程序");
}
System.out.println("----写数据开始----");
System.out.println("----写数据结束----");
System.out.println("结束");
}
}
}
修改后
如果 countDownLatch.await(timeout, unit) 判断超时未全部执行完,就遍历线程池 submit 返回的所有 future,未执行完就中断,最后再调用 await() 等待中断的线程执行 countDownLatch.countDown() 完成所有任务,代码如下:
Task.java:
public class Task implements Runnable {
private final int sleepTime;
private final CountDownLatch countDownLatch;
public Task(int sleepTime, CountDownLatch countDownLatch) {
this.sleepTime = sleepTime;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
try {
Thread.sleep(sleepTime == Demo.THREAD_SIZE ? Demo.LONG_RUNNING_THREAD_TIME : Demo.SHORT_RUNNING_THREAD_TIME);
System.out.println("任务 " + sleepTime + " 干完了");
} catch (InterruptedException ie) {
System.out.println("任务 " + sleepTime + " 被中断");
} finally {
countDownLatch.countDown();
}
}
}
Demo.java:
public class Demo {
/**
* 线程数量
* 前三个任务执行 {@link Demo#SHORT_RUNNING_THREAD_TIME} ms
* 最后一个线程运行 {@link Demo#LONG_RUNNING_THREAD_TIME} ms
*/
public static final int THREAD_SIZE = 4;
/**
* 线程超时时间(ms)
*/
public static final int THREAD_TIMEOUT = 3000;
/**
* 超时线程运行的时间(ms)
*/
public static final int LONG_RUNNING_THREAD_TIME = 20000;
/**
* 正常线程运行的时间(ms)
*/
public static final int SHORT_RUNNING_THREAD_TIME = 10;
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(1);
while (true) {
List<Future<Boolean>> futures = new ArrayList<>(THREAD_SIZE);
CountDownLatch countDownLatch = new CountDownLatch(THREAD_SIZE);
System.out.println("开始");
for (int i = 1; i <= THREAD_SIZE; i++) {
futures.add(executorService.submit(new Task(i, countDownLatch), true));
}
if (!countDownLatch.await(Demo.THREAD_TIMEOUT, TimeUnit.MILLISECONDS)) {
for (Future<Boolean> future : futures) {
if (!future.isDone()) {
future.cancel(true);
}
}
countDownLatch.await();
}
System.out.println("----写数据开始----");
System.out.println("----写数据结束----");
System.out.println("结束");
}
}
}
输出结果:
开始
任务 3 干完了
任务 1 干完了
任务 2 干完了
任务 4 被中断
----写数据开始----
----写数据结束----
结束
开始
任务 3 干完了
任务 1 干完了
任务 2 干完了