线程重用——线程池的基本原理

为简单起见，线程池中只有一个线程：

[java] view plain copy

print?

1. package com.xs.concurrent;  
2.   
3. import java.util.concurrent.BlockingQueue;  
4. import java.util.concurrent.LinkedBlockingQueue;  
5.   
6. public class ThreadPool {  
7.       
8.     private final BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<Runnable>();  
9.       
10.     private final Thread thread;  
11.       
12.     public ThreadPool() {  
13.         thread = ThreadFactory.newThread(blockingQueue);  
14.         thread.start();  
15.     }  
16.       
17.     public void execute(Runnable runnable){  
18.         try {  
19.             blockingQueue.put(runnable);  
20.         } catch (InterruptedException e) {  
21.             throw new RuntimeException(e);  
22.         }  
23.     }  
24.       
25.     public static void main(String[] args) throws Exception {  
26.         ThreadPool pool = new ThreadPool();  
27.         Runnable runnable = new Runnable() {  
28.               
29.             @Override  
30.             public void run() {  
31.                 System.out.println("Hello world!");  
32.             }  
33.         };  
34.         for (int i = 0; i < 10; i++) {  
35.             pool.execute(runnable);  
36.             Thread.sleep(1000);  
37.         }  
38.     }  
39. }  

ThreadFactory：

[java] view plain copy

print?

1. package com.xs.concurrent;  
2.   
3. import java.util.concurrent.BlockingQueue;  
4.   
5. public class ThreadFactory {  
6.       
7.     public static Thread newThread(final BlockingQueue<Runnable> tasks){  
8.         Thread t = new Thread(new Runnable() {  
9.               
10.             @Override  
11.             public void run() {  
12.                 for (;;) {  
13.                     try {  
14.                         Runnable task = tasks.take(); // 阻塞方法，直到取到任务为止  
15.                         task.run();  
16.                     } catch (InterruptedException e) {  
17.                         throw new RuntimeException(e);  
18.                     }  
19.                 }  
20.             }  
21.         });  
22.         return t;  
23.     }  
24.       
25.     private ThreadFactory(){}  
26. }  

Thread.start()不能重复调用，所以要重用Thread，就不能让Thread执行完一个任务后终止，因此就必须阻塞Thread.run()方法，让该方法不停地从任务队列中获取任务并执行。

下面看看JDK的源码。

ThreadPoolExecutor.execute()方法：

[java] view plain copy

print?

1. public void execute(Runnable command) {  
2.         if (command == null)  
3.             throw new NullPointerException();  
4.         if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  
5.             if (runState == RUNNING && workQueue.offer(command)) {  
6.                 if (runState != RUNNING || poolSize == 0)  
7.                     ensureQueuedTaskHandled(command);  
8.             }  
9.             else if (!addIfUnderMaximumPoolSize(command))  
10.                 reject(command); // is shutdown or saturated  
11.         }  
12.     }  

addIfUnderCorePoolSize()：

[java] view plain copy

print?

1. private boolean addIfUnderCorePoolSize(Runnable firstTask) {  
2.         Thread t = null;  
3.         final ReentrantLock mainLock = this.mainLock;  
4.         mainLock.lock();  
5.         try {  
6.             if (poolSize < corePoolSize && runState == RUNNING)  
7.                 t = addThread(firstTask);  
8.         } finally {  
9.             mainLock.unlock();  
10.         }  
11.         if (t == null)  
12.             return false;  
13.         t.start();  
14.         return true;  
15.     }  

这个方法启动了一个线程。

addThread()：

[java] view plain copy

print?

1. private Thread addThread(Runnable firstTask) {  
2.         Worker w = new Worker(firstTask);  
3.         Thread t = threadFactory.newThread(w);  
4.         if (t != null) {  
5.             w.thread = t;  
6.             workers.add(w);  
7.             int nt = ++poolSize;  
8.             if (nt > largestPoolSize)  
9.                 largestPoolSize = nt;  
10.         }  
11.         return t;  
12.     }  

这个方法将任务交给了Worker。Worker是ThreadPoolExecutor的内部类。

Worker.run()：

[java] view plain copy

print?

1. public void run() {  
2.             try {  
3.                 Runnable task = firstTask;  
4.                 firstTask = null;  
5.                 while (task != null || (task = getTask()) != null) {  
6.                     runTask(task); // 执行任务  
7.                     task = null;  
8.                 }  
9.             } finally {  
10.                 workerDone(this);  
11.             }  
12.         }  

getTask()方法属于ThreadPoolExecutor类：

[java] view plain copy

print?

1. Runnable getTask() {  
2.         for (;;) {  
3.             try {  
4.                 int state = runState;  
5.                 if (state > SHUTDOWN)  
6.                     return null;  
7.                 Runnable r;  
8.                 if (state == SHUTDOWN)  // Help drain queue  
9.                     r = workQueue.poll();  
10.                 else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  
11.                     r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);  
12.                 else  
13.                     r = workQueue.take();  
14.                 if (r != null)  
15.                     return r;  
16.                 if (workerCanExit()) {  
17.                     if (runState >= SHUTDOWN) // Wake up others  
18.                         interruptIdleWorkers();  
19.                     return null;  
20.                 }  
21.                 // Else retry  
22.             } catch (InterruptedException ie) {  
23.                 // On interruption, re-check runState  
24.             }  
25.         }  

该方法不停地从工作队列中获取任务。