之前在閱讀《Java并發(fā)編程的藝術(shù)》一書中看到了其中介紹了有關(guān)Java線程池的技術(shù)鼎兽，為了加深自己的理解夜郁，我打算通過書中的一個(gè)簡(jiǎn)單的線程池實(shí)例來強(qiáng)化一下儒将。
在服務(wù)端編程時(shí)，我們常常會(huì)碰到短時(shí)間并且大量的請(qǐng)求捺宗，在一般的情況下柱蟀，我們會(huì)對(duì)每一個(gè)請(qǐng)求都創(chuàng)建一個(gè)線程進(jìn)行處理，然后處理完請(qǐng)求后再關(guān)閉線程蚜厉。這樣看似沒有問題长已，但是假如對(duì)每一個(gè)任務(wù)都創(chuàng)建一個(gè)線程去處理，假如創(chuàng)建的線程很多，系統(tǒng)就會(huì)頻繁地進(jìn)行上下文切換术瓮，這樣就會(huì)加大系統(tǒng)的負(fù)載康聂。
線程池技術(shù)應(yīng)對(duì)上述問題的做法是這樣的：創(chuàng)建一組固定數(shù)量的線程構(gòu)成一個(gè)線程池，將客戶端的請(qǐng)求放到請(qǐng)求隊(duì)列中胞四，然后啟動(dòng)線程池的線程恬汁，其中的線程就會(huì)去請(qǐng)求隊(duì)列獲取需要被執(zhí)行的任務(wù)，問題的關(guān)鍵是在執(zhí)行完任務(wù)之后撬讽，線程并不被關(guān)閉蕊连，而是繼續(xù)從請(qǐng)求隊(duì)列獲取任務(wù)并執(zhí)行悬垃，如果請(qǐng)求隊(duì)列沒有任務(wù)等待被執(zhí)行游昼，那么該線程就處于等待的狀態(tài)。線程池的好處是可以減少后續(xù)創(chuàng)建線程的開銷尝蠕，并能使得系統(tǒng)在固定線程的情況下穩(wěn)定地工作烘豌，我下面使用一個(gè)例子來進(jìn)行一下說明
先定義一個(gè)線程池的接口，里面包含線程池的基本操作看彼，如下所示：
public interface ThreadPool<Job extends Runnable> {
//執(zhí)行一個(gè)Job,這個(gè)Job需要實(shí)現(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();
}
從上面可以看到廊佩，一個(gè)線程池接口包含的功能為
- 將任務(wù)添加到任務(wù)隊(duì)列等待被執(zhí)行的方法
execute(Job job) - 增加和刪除線程池中的線程數(shù)方法
addWorkers(int num)和removeWorkers(int num) - 關(guān)閉線程池中的所有線程方法
shutdown() - 獲取任務(wù)隊(duì)列中的任務(wù)數(shù)方法
getSize()
針對(duì)這個(gè)線程池的接口，我們來對(duì)它進(jìn)行實(shí)現(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ù)隊(duì)列
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;
//線程編號(hào)
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ù)隊(duì)列中靖榕，假如線程池中有空閑的線程标锄，那么該空閑的線程就會(huì)執(zhí)行任務(wù)隊(duì)列里的任務(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ù)隊(duì)列的大小
@Override
public int getSize() {
return jobs.size();
}
//根據(jù)要求創(chuàng)建的線程數(shù)創(chuàng)建一個(gè)特定大小的線程池，并啟動(dò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;
}
}
}
下述是啟動(dòng)代碼:
public class ThreadPoolTest {
public static void main(String[] args){
//創(chuàng)建一個(gè)線程規(guī)模為8的線程池并啟動(dòng)線程池中的所有線程
DefaultThreadPool pool = new DefaultThreadPool(8);
pool.execute(new Job());
}
}
針對(duì)上述代碼茁计，我們做一下分析:
- 開始啟動(dòng)的時(shí)候料皇，程序調(diào)用類
DefaultThreadPool中的DefaultThreadPool(int num)方法來創(chuàng)建線程個(gè)數(shù)為8的線程池，并啟動(dòng)所有的線程 - 被啟動(dòng)的線程獲取等待隊(duì)列
jobs的鎖并進(jìn)入同步隊(duì)列的循環(huán)之中星压，此時(shí)假設(shè)jobs任務(wù)隊(duì)列還沒有被添加任何任務(wù)践剂，那么線程便會(huì)執(zhí)行到jobs.wait()這一行，進(jìn)入休眠狀態(tài)(實(shí)際上是被放到了休眠的隊(duì)列)并釋放已獲得的jobs的鎖以便讓其他的被阻塞在同步塊synchronized (jobs){這里的線程獲取鎖并進(jìn)入同步塊娜膘，但是其它的線程很快發(fā)現(xiàn)自己和第一個(gè)獲取鎖的線程所面對(duì)的狀況一樣逊脯，那么其它線程會(huì)自動(dòng)地添加到正在休眠的線程隊(duì)列后面 - 終于，有一個(gè)任務(wù)通過調(diào)用
pool.execute(new Job())被添加到了任務(wù)隊(duì)列之中竣贪，這個(gè)任務(wù)首先通過獲取jobs的鎖军洼，并把新任務(wù)放到jobs隊(duì)列之中，然后通過調(diào)用jobs.notify()來喚醒睡眠隊(duì)列中因?yàn)槿蝿?wù)隊(duì)列jobs為空而被阻塞的第一個(gè)線程 - 所有沒有被阻塞的線程現(xiàn)在競(jìng)爭(zhēng)任務(wù)隊(duì)列中任務(wù)演怎，假如任務(wù)隊(duì)列中一直存在任務(wù)歉眷，那么這些工作線程在處理完所有的請(qǐng)求后會(huì)再度進(jìn)入睡眠的隊(duì)列
