池化是常見的思想，執行緒池是非常典型的池化的實現，《Java併發程式設計實戰》也大篇幅去講解了Java中的執行緒池。本文實現一個簡單的執行緒池。

2 核心類

【1】介面定義

public interface IThreadPool<Job extends Runnable> {
	/**
	 * 關閉執行緒池
	 */
	public void shutAlldown();

	/**
	 * 執行任務
	 * 
	 * @param job 任務
	 */
	public void execute(Job job);

	/**
	 * 新增工作者
	 * 
	 * @param addNum 新增數
	 */
	public void addWorkers(int addNum);

	/**
	 * 減少工作者
	 * 
	 * @param reduceNum 減少數目
	 */
	public void reduceWorkers(int reduceNum);
}
 

執行緒池的核心是維護了1個任務列表和1個工作者列表。

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

public class XYThreadPool<Job extends Runnable> implements IThreadPool<Job> {

	// 預設執行緒數
	private static int DEAFAULT_SIZE = 5;
	// 最大執行緒數
	private static int MAX_SIZE = 10;

	// 任務列表
	private LinkedList<Job> tasks = new LinkedList<Job>();
	// 工作執行緒列表
	private List<Worker> workers = Collections
			.synchronizedList(new ArrayList<Worker>());

	/**
	 * 預設建構函式
	 */
	public XYThreadPool() {
		initWokers(DEAFAULT_SIZE);
	}

	/**
	 * 執行執行緒數
	 * 
	 * @param threadNums 執行緒數
	 */
	public XYThreadPool(int workerNum) {
		workerNum = workerNum <= 0 ? DEAFAULT_SIZE
				: workerNum > MAX_SIZE ? MAX_SIZE : workerNum;
		initWokers(workerNum);
	}

	/**
	 * 初始化執行緒池
	 * 
	 * @param threadNums 執行緒數
	 */
	public void initWokers(int threadNums) {
		for (int i = 0; i < threadNums; i++) {
			Worker worker = new Worker();
			worker.start();
			workers.add(worker);
		}
		// 新增關閉鉤子
		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				shutAlldown();
			}
		});
	}

	@Override
	public void shutAlldown() {
		for (Worker worker : workers) {
			worker.shutdown();
		}
	}

	@Override
	public void execute(Job job) {
		synchronized (tasks) {
			// 提交任務就是將任務物件加入任務佇列，等待工作執行緒去處理
			tasks.addLast(job);
			tasks.notifyAll();
		}
	}

	@Override
	public void addWorkers(int addNum) {
		// 新執行緒數必須大於零，並且執行緒總數不能大於最大執行緒數
		if ((workers.size() + addNum) <= MAX_SIZE && addNum > 0) {
			initWokers(addNum);
		} else {
			System.out.println("addNum too large");
		}
	}

	@Override
	public void reduceWorkers(int reduceNum) {
		if ((workers.size() - reduceNum <= 0))
			System.out.println("thread num too small");
		else {
			// 暫停指定數量的工作者
			int count = 0;
			while (count != reduceNum) {
				for (Worker w : workers) {
					w.shutdown();
					count++;
				}
			}
		}
	}

	/**
	 * 工作執行緒
	 */
	class Worker extends Thread {

		private volatile boolean flag = true;

		@Override
		public void run() {
			while (flag) {
				Job job = null;
				// 加鎖（若只有一個woker可不必加鎖，那就是所謂的單執行緒的執行緒池，執行緒安全）
				synchronized (tasks) {
					// 任務佇列為空
					while (tasks.isEmpty()) {
						try {
							// 阻塞，放棄物件鎖，等待被notify喚醒
							tasks.wait();
							System.out.println("block when tasks is empty");
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					// 不為空取出任務
					job = tasks.removeFirst();
					System.out.println("get job:" + job + ",do biz");
					job.run();
				}
			}
		}

		public void shutdown() {
			flag = false;
		}
	}
}
 

1 當呼叫wait()方法時執行緒會放棄物件鎖，進入等待此物件的等待鎖定池，只有針對此物件呼叫notify()方法後本執行緒才進入物件鎖定池準備

2 Object的方法：void notify(): 喚醒一個正在等待該物件的執行緒。void notifyAll(): 喚醒所有正在等待該物件的執行緒。notifyAll使所有原來在該物件上等待被notify的執行緒統統退出wait狀態，變成等待該物件上的鎖，一旦該物件被解鎖，它們會去競爭。notify只是選擇一個wait狀態執行緒進行通知，並使它獲得該物件上的鎖，但不驚動其它同樣在等待被該物件notify的執行緒們，當第一個執行緒執行完畢以後釋放物件上的鎖，此時如果該物件沒有再次使用notify語句，即便該物件已經空閒，其他wait狀態等待的執行緒由於沒有得到該物件的通知，繼續處在wait狀態，直到這個物件發出一個notify或notifyAll，

public class TestService1 {
	public static void main(String[] args) {
		// 啟動10個執行緒
		XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
		pool.execute(new Runnable() {
			@Override
			public void run() {
				System.out.println("====1 test====");
			}
		});	
	}
}

public class TestService2 {
	public static void main(String[] args) {
		// 啟動10個執行緒
		XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
		pool.execute(new Runnable() {
			@Override
			public void run() {
				System.out.println("====2 test====");
			}
		});
	}
}

import javax.annotation.PostConstruct;
import org.springframework.stereotype.Component;
import com.xy.pool.XYThreadPool;

/**
 * 統一執行緒池管理類 
 */
@Component
public class XYThreadManager {

	private XYThreadPool<Runnable> executorPool;

	@PostConstruct
	public void init() {
		executorPool = new XYThreadPool<Runnable>(10);
	}

	public XYThreadPool<Runnable> getExecutorPool() {
		return executorPool;
	}
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service("testService3")
public class TestService3 {
	
	@Autowired
	private XYThreadManager threadManager;
	
	public void test() {
		threadManager.getExecutorPool().execute(new Runnable() {
			@Override
			public void run() {
				System.out.println("====3 test====");
			}
		});
	}
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service("testService4")
public class TestService4 {
	
	@Autowired
	private XYThreadManager threadManager;
	
	public void test() {
		threadManager.getExecutorPool().execute(new Runnable() {
			@Override
			public void run() {
				System.out.println("====4 test====");
			}
		});
	}
}

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class TestMain {

	@SuppressWarnings("resource")
	public static void main(String[] args) {
		ApplicationContext atc = new ClassPathXmlApplicationContext("applicationContext.xml");

		TestService3 t3 = (TestService3) atc.getBean("testService3");
		t3.test();

		TestService4 t4 = (TestService4) atc.getBean("testService4");
		t4.test();
	}

}