云智慧（北京）科技有限公司 陈鑫
场景
一个调度器，两个调度任务，分别处理两个目录下的txt文件，某个调度任务应对某些复杂问题的时候会持续特别长的时间，甚至有一直阻塞的可能。我们需要一个manager来管理这些task，当这个task的上一次执行时间距离现在超过5个调度周期的时候，就直接停掉这个线程，然后再重启它，保证两个目标目录下没有待处理的txt文件堆积。
问题
直接使用java默认的线程池调度task1和task2.由于外部txt的种种不可控原因，导致task2线程阻塞。现象就是task1和线程池调度器都正常运行着，但是task2迟迟没有动作。
当然，找到具体的阻塞原因并进行针对性解决是很重要的。但是，这种措施很可能并不能完全、彻底、全面的处理好所有未知情况。我们需要保证任务线程或者调度器的健壮性！
方案计划
线程池调度器并没有原生的针对被调度线程的业务运行状态进行监控处理的API。因为task2是阻塞在我们的业务逻辑里的，所以最好的方式是写一个TaskManager，所有的任务线程在执行任务前全部到这个TaskManager这里来注册自己。这个TaskManager就负责对于每个自己管辖范围内的task进行实时全程监控！
后面的重点就是如何处理超过5个执行周期的task了。
方案如下：
一旦发现这个task线程，立即中止它，然后再次重启；
一旦发现这个task线程，直接将整个pool清空并停止，重新放入这两个task ——【task明确的情况下】；
方案实施
中止后重启
Task实现类
classFileTask extends Thread {
private long lastExecTime = 0;
protected long interval = 10000;
public long getLastExecTime() {
returnlastExecTime;
}
public void setLastExecTime(longlastExecTime) {
this.lastExecTime =lastExecTime;
}
public long getInterval() {
return interval;
}
public void setInterval(long interval) {
this.interval = interval;
}
public File[] getFiles() {
return null;
}
@Override
public void run() {
while(!Thread.currentThread().isInterrupted()) {
lastExecTime = System.currentTimeMillis();
System.out.println(Thread.currentThread().getName() + " is running ->" + new Date());
try {
Thread.sleep(getInterval() * 6 * 1000);
} catch(InterruptedException e) {
Thread.currentThread().interrupt();
e.printStackTrace(); // 当线程池shutdown之后，这里就会抛出exception了
}
}
}
}
TaskManager
public class TaskManager implements Runnable {
private final static Log logger = LogFactory.getLog(TaskManager.class);
public Set<FileTask> runners = newCopyOnWriteArraySet<FileTask>();
ExecutorService pool =Executors.newCachedThreadPool();
public voidregisterCodeRunnable(FileTask process) {
runners.add(process);
}
publicTaskManager (Set<FileTask>runners) {
this.runners = runners;
}
@Override
public void run() {
while(!Thread.currentThread().isInterrupted()) {
try {
long current = System.currentTimeMillis();
for (FileTask wrapper : runners) {
if (current - wrapper.getLastExecTime() >wrapper.getInterval()* 5) {
wrapper.interrupt();
for (File file : wrapper.getFiles()) {
file.delete();
}
wrapper.start();
}
}
} catch(Exception e1) {
logger.error("Error happens when we trying to interrupt and restart a task");
ExceptionCollector.registerException(e1);
}
try {
Thread.sleep(500);
} catch(InterruptedException e) {
}
}
}
}
这段代码会报错java.lang.Thread IllegalThreadStateException。为什么呢？其实这是一个很基础的问题，您应该不会像我一样马虎。查看Thread.start()的注释, 有这样一段：
It is never legal to start a thread more thanonce. In particular, a thread may not be restarted once it has completedexecution.