場景

一個調度器，兩個調度任務，分別處理兩個目錄下的txt文件，某個調度任務應對某些復雜問題的時候會持續特別長的時間，甚至有一直阻塞的可能。我們需要一個manager來管理這些task，當這個task的上一次執行時間距離現在超過5個調度周期的時候，就直接停掉這個線程，然后再重啟它，保證兩個目標目錄下沒有待處理的txt文件堆積。

問題

直接使用java默認的線程池調度task1和task2.由于外部txt的種種不可控原因，導致task2線程阻塞?，F象就是task1和線程池調度器都正常運行著，但是task2遲遲沒有動作。

當然，找到具體的阻塞原因并進行針對性解決是很重要的。但是，這種措施很可能并不能完全、徹底、全面的處理好所有未知情況。我們需要保證任務線程或者調度器的健壯性!

方案計劃

線程池調度器并沒有原生的針對被調度線程的業務運行狀態進行監控處理的API。因為task2是阻塞在我們的業務邏輯里的，所以***的方式是寫一個TaskManager，所有的任務線程在執行任務前全部到這個TaskManager這里來注冊自己。這個TaskManager就負責對于每個自己管轄范圍內的task進行實時全程監控!

后面的重點就是如何處理超過5個執行周期的task了。

方案如下：

●一旦發現這個task線程，立即中止它，然后再次重啟;

●一旦發現這個task線程，直接將整個pool清空并停止，重新放入這兩個task ——【task明確的情況下】;

方案實施

中止后重啟

●Task實現類

class FileTask extends Thread { 
private long lastExecTime = 0; 
protected long interval = 10000; 
public long getLastExecTime() { 
    return lastExecTime; 
} 
public void setLastExecTime(long lastExecTime) { 
    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 = new CopyOnWriteArraySet<FileTask>(); 
ExecutorService pool = Executors.newCachedThreadPool(); 
public void registerCodeRunnable(FileTask process) { 
runners.add(process); 
} 
public TaskManager (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 than once. In particular, a thread may not be restarted once it has completed execution.

是的，一個線程不能夠啟動兩次。那么它是怎么判斷的呢?

public synchronized void start() { 
        /** 
         * A zero status value corresponds to state "NEW".    0對應的是state NEW 
         */ 

if (threadStatus != 0) //如果不是NEW state，就直接拋出異常!#p#

throw new IllegalThreadStateException(); 
        group.add(this); 
        boolean started = false; 
        try { 
        start0();    // 啟動線程的native方法 
        started = true; 
        } finally { 
            try { 
                if (!started) { 
                    group.threadStartFailed(this); 
                } 
            } catch (Throwable ignore) { 
            } 
        } 
    } 

恩，只有是NEW狀態才能夠調用native方法啟動一個線程。好吧，到這里了，就普及也自補一下jvm里的線程狀態：

所有的線程狀態：:

●NEW —— 還沒有啟動過

●RUNNABLE —— 正在jvm上運行著

●BLOCKED —— 正在等待鎖/信號量被釋放

●WAITING —— 等待其他某個線程的某個特定動作

●TIMED_WAITING —— A thread that is waiting for another thread to perform an action for up to a specified waiting time is in this state.

●TERMINATED —— 退出，停止

線程在某個時間點上只可能存在一種狀態，這些狀態是jvm里的，并不反映操作系統線程的狀態。查一下Thread的API，沒有對其狀態進行修改的API。那么這條路是不通的嗎?

仔細考慮一下……

如果把任務做成Runnable實現類，然后在把這個實現類丟進線程池調度器之前，利用此Runnable構造一個Thread，是不是這個Thread對象就能夠控制這個runnable對象，進而控制在線程池中運行著的task了呢?非也!讓我們看看Thread和ThreadPoolExecutor對Runnable的處理吧。

●Thread

/* What will be run. */ 
private Runnable target; 

結合上面的start()方法，很容易猜出，start0()會把target弄成一個線程來進行運行。

●ThreadPoolExecutor

public void execute(Runnable command) { 
        if (command == null) 
            throw new NullPointerException(); 
        int c = ctl.get(); 
        if (workerCountOf(c) < corePoolSize) { 
            if (addWorker(command, true)) 
                return; 
            c = ctl.get(); 
        } 
        if (isRunning(c) && workQueue.offer(command)) { 
            int recheck = ctl.get(); 
            if (! isRunning(recheck) && remove(command)) 
                reject(command); 
            else if (workerCountOf(recheck) == 0) 
                addWorker(null, false); 
        } 
        else if (!addWorker(command, false)) 
            reject(command); 
} 
private boolean addWorker(Runnable firstTask, boolean core) { 
… 
boolean workerStarted = false; 
boolean workerAdded = false; 
Worker w = null; 
try { 
final ReentrantLock mainLock = this.mainLock; 
w = new Worker(firstTask); 
final Thread t = w.thread; 
if (t != null) { 
mainLock.lock(); 
try { 
int c = ctl.get(); 
int rs = runStateOf(c); 
if (rs < SHUTDOWN || 
(rs == SHUTDOWN && firstTask == null)) { 
if (t.isAlive()) // precheck that t is startable 
throw new IllegalThreadStateException(); 
workers.add(w); 
int s = workers.size(); 
if (s > largestPoolSize) 
largestPoolSize = s; 
workerAdded = true; 
} 
} finally { 
mainLock.unlock(); 
} 
if (workerAdded) { 
t.start(); 
workerStarted = true; 
} 
} 
} finally { 
if (! workerStarted) 
addWorkerFailed(w); 
} 
return workerStarted; 
} 

那么Worker又是怎樣的呢?

●Worker

private final class Worker 
extends AbstractQueuedSynchronizer 
implements Runnable 
{ 
final Thread thread; 
Runnable firstTask; 
volatile long completedTasks; 
Worker(Runnable firstTask) { 
setState(-1); //調用runWorker之前不可以interrupt 
this.firstTask = firstTask; 
this.thread = getThreadFactory().newThread(this); 
} 
public void run() { 
runWorker(this); 
} 
……   
……. 
void interruptIfStarted() { 
Thread t; 
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { 
try { 
t.interrupt(); 
} catch (SecurityException ignore) { 
} 
} 
} 
} 

可見worker里既包裝了Runnable對象——task，又包裝了一個Thread對象——以自己作為初始化參數，因為worker也是Runnable對象。然后對外提供了運行與停止接口，run()和interruptIfStarted()。回顧上面使用Thread的例子不禁有了新的領悟，我們把一個Thread對象交給ThreadPoolExecutor執行后，實際的調用是對Thread(FileTask())對象，我們暫時稱之為workerWrapper。那么我們在池外進行FileTask.interrupt()操作影響的是FileTask對象，而不是workerWrapper。所以可能上面對于start()方法二次調用不是特別適當。更恰當的應該是在fileTask.interrupt()的時候就跑出異常，因為從來沒有對fileTask對象執行過start()方法，這時候去interrupt就會出現錯誤。具體如下圖：

分析到此，我們已經明確除了調用ThreadPoolExecutor了的interruptWorkers()方法別無其他途徑操作這些worker了。

private void interruptWorkers() { 
final ReentrantLock mainLock = this.mainLock; 
mainLock.lock(); 
try { 
for (Worker w : workers) 
w.interruptIfStarted(); 
} finally { 
mainLock.unlock(); 
} 
} 

重啟線程池

●TaskManager

public class TaskManager  implements Runnable { 
….. 
public TaskManager (Set<FileTask> runners) { 
super(); 
this.runners = runners; 
executeTasks(runners); 
} 
private void executeTasks(Set<FileTask> runners) { 
for (FileTask task : runners) { 
pool.execute(task); 
System.out.println(task.getClass().getSimpleName() + " has been started"); 
} 
} 

@Override

public void run() { 
while (!Thread.currentThread().isInterrupted()) { 
try { 
long current = System.currentTimeMillis(); 
for (FileTask wrapper : runners) { 
if (wrapper.getLastExecTime() != 0 && current - wrapper.getLastExecTime() > wrapper.getInterval() * 5 * 1000) {    // 開始忘了乘以1000 
wrapper.interrupt(); 
if (wrapper.getFiles() != null){ 
for (File file : wrapper.getFiles()) { 
file.delete(); 
} 
} 
System.out.println("Going to shutdown the thread pool"); 
List<Runnable> shutdownNow = pool.shutdownNow();    // 不等當前pool里的任務執行完，直接關閉線程池 
for (Runnable run : shutdownNow) { 
System.out.println(run + " going to be shutdown"); 
} 
while (pool.awaitTermination(1, TimeUnit.SECONDS)) {   
System.out.println("The thread pool has been shutdown " + new Date()); 
executeTasks(runners); // 重新執行 
Thread.sleep(200); 
} 
} 
} 
} catch (Exception e1) { 
e1.printStackTrace(); 
} 
try { 
Thread.sleep(500); 
} catch (InterruptedException e) { 
} 
} 
} 
public static void main(String[] args) { 
Set<FileTask> tasks = new HashSet<FileTask>(); 
        
FileTask task = new FileTask(); 
task.setInterval(1); 
task.setName("task-1"); 
tasks.add(task); 
               
FileTask task1 = new FileTask(); 
task1.setInterval(2); 
task.setName("task-2"); 
tasks.add(task1); 
        
TaskManager  codeManager = new TaskManager (tasks); 
new Thread(codeManager).start(); 
}    
} 

成功!把整個的ThreadPoolExector里所有的worker全部停止，之后再向其隊列里重新加入要執行的兩個task(注意這里并沒有清空，只是停止而已)。這樣做雖然能夠及時處理task，但是一個很致命的缺點在于，如果不能明確的知道ThreadPoolExecutor要執行的task，就沒有辦法重新執行這些任務。#p#

定制線程池

好吧!停止鉆研別人的東西!我們完全可以自己寫一個自己的ThreadPoolExecutor，只要把worker暴露出來就可以了。這里是不是回想起前面的start問題來了，沒錯，我們即便能夠直接針對Thread進行interrupt, 但是不能再次start它了。那么clone一個同樣的Thread行不行呢?#p#

●Thread

@Override 
protected Object clone() throws CloneNotSupportedException { 
throw new CloneNotSupportedException(); 
} 

答案顯而易見，線程是不支持clone 的。我們需要重新new 一個Thread來重新運行。其實我們只需要將原來的Worker里的Runnable換成我們自己的task，然后將訪問權限適當放開就可以了。還有，就是讓我們的CustomThreadPoolExecutor繼承Thread，因為它需要定時監控自己的所有的worker里Thread的運行狀態。

●CustomThreadPoolExecutor

public class CustomThreadPoolExecutor extends ThreadPoolExecutor implements Runnable {  
public void execute(Testask command) { 
….//將執行接口改為接收我們的業務類 
} 
… 
… 
private final class Worker 
extends AbstractQueuedSynchronizer 
implements Runnable 
{ 
… 
Testask firstTask; //將Runnable改為我們的業務類，方便查看狀態 
… 
Worker(Testask firstTask) { 
…//同樣將初始化參數改為我們的業務類 
} 
 
} 
public static void main(String[] args) { 
CustomThreadPoolExecutor pool = new CustomThreadPoolExecutor(0, Integer.MAX_VALUE, 
60L, TimeUnit.SECONDS, 
new SynchronousQueue<Runnable>()); 
         
Testask task = new Testask(); 
task.setInterval(1); 
pool.execute(task); 
         
Testask task1 = new Testask(); 
task1.setInterval(2); 
pool.execute(task1); 
         
new Thread(pool).start(); 
} 
 
@Override 
public void run() { 
while (!Thread.currentThread().isInterrupted()) { 
try { 
long current = System.currentTimeMillis(); 
Set<Testask> toReExecute = new HashSet<Testask>(); 
System.out.println("\t number is " + number); 
for (Worker wrapper : workers) { 
Testask tt = wrapper.firstTask; 
if (tt != null) { 
if (current - tt.getLastExecTime() > tt.getInterval() * 5 * 1000) { 
wrapper.interruptIfStarted(); 
remove(tt); 
if (tt.getFiles() != null) { 
for (File file : tt.getFiles()) { 
file.delete(); 
} 
} 
System.out.println("THread is timeout : " + tt + " " + new Date()); 
toReExecute.add(tt); 
} 
} 
} 
if (toReExecute.size() > 0) { 
mainLock.lock(); 
try { 
for (Testask tt : toReExecute) { 
execute(tt);    // execute this task again 
}  
} finally { 
mainLock.unlock(); 
} 
} 
} catch (Exception e1) { 
System.out.println("Error happens when we trying to interrupt and restart a code task "); 
} 
try { 
Thread.sleep(500); 
} catch (InterruptedException e) { 
} 
} 
} 
} 

●Testask

class Testask implements Runnable { 
….. 
 
@Override 
public void run() { 
while (!Thread.currentThread().isInterrupted()) { 
lastExecTime = System.currentTimeMillis(); 
System.out.println(Thread.currentThread().getName() + " is running -> " + new Date()); 
try { 
CustomThreadPoolExecutor.number++; 
Thread.sleep(getInterval() * 6 * 1000); 
                System.out.println(Thread.currentThread().getName() + " after sleep"); 
} catch (InterruptedException e) { 
Thread.currentThread().interrupt(); 
System.out.println("InterruptedException happens"); 
} 
} 
System.out.println("Going to die"); 
} 
} 

最終方案

綜上，最穩妥的就是使用JDK自帶的ThreadPoolExecutor, 如果需要對池里的task進行任意時間的控制，可以考慮全面更新，全方面，360度無死角的定制自己的線程池當然是***的方案，但是一定要注意對于共享對象的處理，適當的處理好并發訪問共享對象的方法。

鑒于我們的場景，由于時間緊，而且需要了解的task并不多，暫時選用全部重新更新的策略。上線后，抽時間把自己定制的ThreadPoolExecutor搞定，然后更新上去!