之前在閱讀《Java并發(fā)編程的藝術(shù)》一書中看到了其中介紹了有關(guān)Java線程池的技術(shù)，為了加深自己的理解，我打算通過書中的一個簡單的線程池實例來強化一下。
在服務(wù)端編程時，我們常常會碰到短時間并且大量的請求，在一般的情況下，我們會對每一個請求都創(chuàng)建一個線程進(jìn)行處理，然后處理完請求后再關(guān)閉線程。這樣看似沒有問題，但是假如對每一個任務(wù)都創(chuàng)建一個線程去處理，假如創(chuàng)建的線程很多，系統(tǒng)就會頻繁地進(jìn)行上下文切換，這樣就會加大系統(tǒng)的負(fù)載。
線程池技術(shù)應(yīng)對上述問題的做法是這樣的：創(chuàng)建一組固定數(shù)量的線程構(gòu)成一個線程池，將客戶端的請求放到請求隊列中，然后啟動線程池的線程，其中的線程就會去請求隊列獲取需要被執(zhí)行的任務(wù)，問題的關(guān)鍵是在執(zhí)行完任務(wù)之后，線程并不被關(guān)閉，而是繼續(xù)從請求隊列獲取任務(wù)并執(zhí)行，如果請求隊列沒有任務(wù)等待被執(zhí)行，那么該線程就處于等待的狀態(tài)。線程池的好處是可以減少后續(xù)創(chuàng)建線程的開銷，并能使得系統(tǒng)在固定線程的情況下穩(wěn)定地工作，我下面使用一個例子來進(jìn)行一下說明
先定義一個線程池的接口，里面包含線程池的基本操作，如下所示：
public interface ThreadPool<Job extends Runnable> {
//執(zhí)行一個Job,這個Job需要實現(xiàn)Runnabale
public void execute(Job job);
//關(guān)閉線程池
public void shutdown();
//添加工作線程
public void addWorkers(int num);
//移除工作線程
public void removeWorkers(int num);
//得到正在等待執(zhí)行的任務(wù)數(shù)量
public int getSize();
}
從上面可以看到，一個線程池接口包含的功能為
- 將任務(wù)添加到任務(wù)隊列等待被執(zhí)行的方法
execute(Job job) - 增加和刪除線程池中的線程數(shù)方法
addWorkers(int num)和removeWorkers(int num) - 關(guān)閉線程池中的所有線程方法
shutdown() - 獲取任務(wù)隊列中的任務(wù)數(shù)方法
getSize()
針對這個線程池的接口，我們來對它進(jìn)行實現(xiàn)如下:
public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job> {
//設(shè)置工作者數(shù)量上限
private static final int MAX_WORKER_NUMBERS = 10;
//默認(rèn)的工作者數(shù)量
private static final int DEFAULT_WORKER_NUMBERS = 5;
//設(shè)置工作者數(shù)量下限
private static final int MIN_WORKER_NUMBER = 1;
//接收的任務(wù)隊列
private final LinkedList<Job> jobs = new LinkedList<Job>();
//工作者
private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());
//工作者線程的數(shù)量
private int workerNum = DEFAULT_WORKER_NUMBERS;
//線程編號
private AtomicInteger threadnum = new AtomicInteger();
public DefaultThreadPool(){
initializeWorkers(DEFAULT_WORKER_NUMBERS);
}
//創(chuàng)建線程池中的線程數(shù)
public DefaultThreadPool(int num){
int workernum = num > MAX_WORKER_NUMBERS ? MAX_WORKER_NUMBERS : num < MIN_WORKER_NUMBER ? MIN_WORKER_NUMBER : num;
initializeWorkers(workernum);
}
//將任務(wù)添加到任務(wù)隊列中，假如線程池中有空閑的線程，那么該空閑的線程就會執(zhí)行任務(wù)隊列里的任務(wù)
@Override
public void execute(Job job) {
if(job != null){
synchronized (jobs){
jobs.addLast(job);
jobs.notify();
}
}
}
//關(guān)閉線程池中的所有線程
@Override
public void shutdown() {
for(Worker worker : workers)
worker.shutdown();
}
//增加線程池中的線程數(shù)
@Override
public void addWorkers(int num) {
synchronized (jobs){
if(num + this.workerNum > MAX_WORKER_NUMBERS)
num = MAX_WORKER_NUMBERS - this.workerNum;
initializeWorkers(num);
this.workerNum += num;
}
}
//移除線程池中特定數(shù)目的線程
@Override
public void removeWorkers(int num) {
synchronized (jobs){
if(num >= this.workerNum)
throw new IllegalArgumentException("beyond workerNum");
int count = 0;
while (count < num){
Worker worker = workers.get(count);
if(workers.remove(worker)){
worker.shutdown();
count++;
}
}
this.workerNum -= count;
}
}
//獲取任務(wù)隊列的大小
@Override
public int getSize() {
return jobs.size();
}
//根據(jù)要求創(chuàng)建的線程數(shù)創(chuàng)建一個特定大小的線程池，并啟動線程池中的所有線程
public void initializeWorkers(int num){
for(int i = 0;i < num;i++){
Worker worker = new Worker();
workers.add(worker);
Thread thread = new Thread(worker,"Thread-num-" + threadnum.incrementAndGet());
thread.start();
}
}
class Worker implements Runnable{
private volatile boolean running = true;
@Override
public void run() {
while (running){
Job job = null;
synchronized (jobs){
while (jobs.isEmpty()){
try {
jobs.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
job = jobs.removeFirst();
}
if(job != null){
job.run();
}
}
}
public void shutdown(){
running = false;
}
}
}
下述是啟動代碼:
public class ThreadPoolTest {
public static void main(String[] args){
//創(chuàng)建一個線程規(guī)模為8的線程池并啟動線程池中的所有線程
DefaultThreadPool pool = new DefaultThreadPool(8);
pool.execute(new Job());
}
}
針對上述代碼，我們做一下分析:
- 開始啟動的時候，程序調(diào)用類
DefaultThreadPool中的DefaultThreadPool(int num)方法來創(chuàng)建線程個數(shù)為8的線程池，并啟動所有的線程 - 被啟動的線程獲取等待隊列
jobs的鎖并進(jìn)入同步隊列的循環(huán)之中，此時假設(shè)jobs任務(wù)隊列還沒有被添加任何任務(wù)，那么線程便會執(zhí)行到jobs.wait()這一行，進(jìn)入休眠狀態(tài)(實際上是被放到了休眠的隊列)并釋放已獲得的jobs的鎖以便讓其他的被阻塞在同步塊synchronized (jobs){這里的線程獲取鎖并進(jìn)入同步塊，但是其它的線程很快發(fā)現(xiàn)自己和第一個獲取鎖的線程所面對的狀況一樣，那么其它線程會自動地添加到正在休眠的線程隊列后面 - 終于，有一個任務(wù)通過調(diào)用
pool.execute(new Job())被添加到了任務(wù)隊列之中，這個任務(wù)首先通過獲取jobs的鎖，并把新任務(wù)放到jobs隊列之中，然后通過調(diào)用jobs.notify()來喚醒睡眠隊列中因為任務(wù)隊列jobs為空而被阻塞的第一個線程 - 所有沒有被阻塞的線程現(xiàn)在競爭任務(wù)隊列中任務(wù)，假如任務(wù)隊列中一直存在任務(wù)，那么這些工作線程在處理完所有的請求后會再度進(jìn)入睡眠的隊列
