前段時間自己研究了下線程池的實現(xiàn)原理，通過一些源碼對比，發(fā)現(xiàn)其實核心的東西不難，于是抽絲剝繭，決定自己實現(xiàn)一個簡單線程池，當(dāng)自已實現(xiàn)了出一個線程池后。發(fā)現(xiàn)原來那么高大上的東西也可以這么簡單。
先上原理圖：為了更好的在手機上顯示，我重新把圖畫了一遍
上代碼之前，要先補充一下線程池構(gòu)造的核心幾個點
- 線程池里的核心線程數(shù)與最大線程數(shù)
- 線程池里真正工作的線程
worker - 線程池里用來存取任務(wù)的隊列
BlockingQueue - 線程中的任務(wù)
task
本例實現(xiàn)簡化了一些，只實現(xiàn)了BlockingQueue存放任務(wù)，然后每個worker取任務(wù)并執(zhí)行，下面看代碼
首先定義一個線程池ThreadExcutor
class ThreadExcutor{
//創(chuàng)建
private volatile boolean RUNNING = true;
//所有任務(wù)都放隊列中，讓工作線程來消費
private static BlockingQueue<Runnable> queue = null;
private final HashSet<Worker> workers = new HashSet<Worker>();
private final List<Thread> threadList = new ArrayList<Thread>();
//工作線程數(shù)
int poolSize = 0;
//核心線程數(shù)（創(chuàng)建了多少個工作線程）
int coreSize = 0;
boolean shutdown = false;
public ThreadExcutor(int poolSize){
this.poolSize = poolSize;
queue = new LinkedBlockingQueue<Runnable>(poolSize);
}
public void exec(Runnable runnable) {
if (runnable == null) throw new NullPointerException();
if(coreSize < poolSize){
addThread(runnable);
}else{
//System.out.println("offer" + runnable.toString() + " " + queue.size());
try {
queue.put(runnable);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void addThread(Runnable runnable){
coreSize ++;
Worker worker = new Worker(runnable);
workers.add(worker);
Thread t = new Thread(worker);
threadList.add(t);
try {
t.start();
}catch (Exception e){
e.printStackTrace();
}
}
public void shutdown() {
RUNNING = false;
if(!workers.isEmpty()){
for (Worker worker : workers){
worker.interruptIfIdle();
}
}
shutdown = true;
Thread.currentThread().interrupt();
}
//這里留個位置放內(nèi)部類Worker
}
然后定義一個內(nèi)部類Worker，這個內(nèi)部類Worker是用來執(zhí)行每個任務(wù)的，在創(chuàng)建線程池后，往線程里添加任務(wù)，每個任務(wù)都是由Worker一個一個來啟動的。
/**
* 工作線程
*/
class Worker implements Runnable{
public Worker(Runnable runnable){
queue.offer(runnable);
}
@Override
public void run() {
while (true && RUNNING){
if(shutdown == true){
Thread.interrupted();
}
Runnable task = null;
try {
task = getTask();
task.run();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public Runnable getTask() throws InterruptedException {
return queue.take();
}
public void interruptIfIdle() {
for (Thread thread :threadList) {
System.out.println(thread.getName() + " interrupt");
thread.interrupt();
}
}
}
首先注意的一點，這個Worker是個內(nèi)部類，是在線程池內(nèi)聲明的。
exec方法
Worker怎么工作
這個工作線程實例化的時候就先加入一個任務(wù)到隊列中，也就是說在實例化這個工作線程時，這個工作線程也是一個任務(wù)被加入到線程池中。然后就是run方法，這個run方法是線程調(diào)start方法生成的線程，而Worker調(diào)的run方法并沒有生成新的線程。就是一個循環(huán)，一直在不停的從隊列中取任務(wù)，然后執(zhí)行。可以看到，取隊列的方法是take()，這個方法意思如果隊列為空了，取不到數(shù)據(jù)時就阻塞隊列。
然后看shutdown()
你每天辛勤的勞動著，突然接收到上面的命令，說活暫時不要接了，先停下來，當(dāng)你還沒搞清楚狀況時，接著你的領(lǐng)導(dǎo)又把你開除了，說公司要倒了，你先下崗吧，一會我也得下崗了。這就是shutdown做的事，shutdown必須是主線程才能停止工作線程。
shutdown方法并不是用線程那種強制停止的搞法，而是先用一個標識符告訴工作線程，不要再接任務(wù)了。然后通知工作線程，你可以interrupt()了，當(dāng)所有的線程停止后記得要把主線程也停掉，這樣，一個簡單任務(wù)的線程池就完成了。
讓我們來測試一下：
/**
* Created by wxwall on 2017/6/7.
*/
public class TheadBlockedQ {
public static void main(String[] args) throws InterruptedException {
ThreadExcutor excutor = new ThreadExcutor(3);
for (int i = 0; i < 10; i++) {
excutor.exec(new Runnable() {
@Override
public void run() {
System.out.println("線程 " + Thread.currentThread().getName() + " 在幫我干活");
}
});
}
excutor.shutdown();
}
}
輸出結(jié)果為：
線程 Thread-0 在幫我干活
線程 Thread-2 在幫我干活
線程 Thread-1 在幫我干活
線程 Thread-0 在幫我干活
線程 Thread-2 在幫我干活
線程 Thread-2 在幫我干活
線程 Thread-1 在幫我干活
線程 Thread-0 在幫我干活
Thread-0 interrupt
Thread-1 interrupt
Thread-2 interrupt
Thread-0 interrupt
Thread-1 interrupt
Thread-2 interrupt
Thread-0 interrupt
Thread-1 interrupt
Thread-2 interrupt
這當(dāng)然是最簡單實現(xiàn)，JDK的實現(xiàn)比這強大的多，而且還具備當(dāng)工作線程處理不過來時，可以產(chǎn)生新的線程來處理任務(wù)，這個數(shù)量不能超過原先定義的最大線程數(shù)，而在本例中都沒實現(xiàn)這些功能。
我相信當(dāng)想了解一個模塊的功能時，如果一開始就了解其中最核心的點，然后向外慢慢擴展，那么學(xué)習(xí)這個模塊時一定能省下不少時間，而且理解將很深刻。希望這個簡單線程池實現(xiàn)能讓你有所領(lǐng)悟，以更加簡單的方式了解線程池，了解了線程池，對于其他池化技術(shù)，原理都是相通的。
最后我想說：我相信寫好一篇文章能讓大家理解不困難都是用了心的，您也點個贊支持支持。
