學(xué)生視角手把手帶你寫(xiě)Java?線(xiàn)程池初版

更新時(shí)間：2022年03月21日 11:16:08 作者：摸魚(yú)打醬油

作者是一個(gè)來(lái)自河源的大三在校生，以下筆記都是作者自學(xué)之路的一些淺薄經(jīng)驗(yàn)，如有錯(cuò)誤請(qǐng)指正，將來(lái)會(huì)不斷的完善筆記，幫助更多的Java愛(ài)好者入門(mén)

手寫(xiě)線(xiàn)程池-定義參數(shù)
手寫(xiě)線(xiàn)程池-構(gòu)造器
手寫(xiě)線(xiàn)程池-默認(rèn)構(gòu)造器
手寫(xiě)線(xiàn)程池-execute方法
手寫(xiě)線(xiàn)程池-處理任務(wù)
手寫(xiě)線(xiàn)程池-優(yōu)雅關(guān)閉線(xiàn)程池
手寫(xiě)線(xiàn)程池-暴力關(guān)閉線(xiàn)程池
手寫(xiě)線(xiàn)程池-源代碼
問(wèn)題

Java手寫(xiě)線(xiàn)程池（第一代）

經(jīng)常使用線(xiàn)程池，故今天突發(fā)奇想，手寫(xiě)一個(gè)線(xiàn)程池，會(huì)有很多不足，請(qǐng)多多寬容。因?yàn)檫@也是第一代的版本，后續(xù)會(huì)更完善。

手寫(xiě)線(xiàn)程池-定義參數(shù)

	private final AtomicInteger taskcount=new AtomicInteger(0);
    private final AtomicInteger threadNumber=new AtomicInteger(0);
    private volatile int corePoolSize; 
    private final Set<MyThreadPoolExecutor.MyWorker> workers; 
    private final BlockingQueue<Runnable> waitingQueue; 
    private final String THREADPOOL_NAME="MyThread-Pool-";
    private volatile boolean isRunning=true; 
    private volatile boolean STOPNOW=false; 
    private final ThreadFactory threadFactory;

taskcount：執(zhí)行任務(wù)次數(shù)
threadNumber：線(xiàn)程編號(hào)，從0開(kāi)始依次遞增。
corePoolSize：核心線(xiàn)程數(shù)
workers：工作線(xiàn)程
waitingQueue：等待隊(duì)列
THREADPOOL_NAME：線(xiàn)程名稱(chēng)
isRunning：是否運(yùn)行
STOPNOW：是否立刻停止
threadFactory：線(xiàn)程工廠

手寫(xiě)線(xiàn)程池-構(gòu)造器

    public MyThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> waitingQueue,ThreadFactory threadFactory) {
        this.corePoolSize=corePoolSize;
        this.workers=new HashSet<>(corePoolSize);
        this.waitingQueue=waitingQueue;
        this.threadFactory=threadFactory;
        //線(xiàn)程預(yù)熱
        for (int i = 0; i < corePoolSize; i++) {
            new MyWorker();
        }
    }

該構(gòu)造器作用：

1：對(duì)參數(shù)進(jìn)行賦值。

2：線(xiàn)程預(yù)熱。根據(jù)corePoolSize的大小來(lái)調(diào)用MyWorker的構(gòu)造器。我們可以看看MyWorker構(gòu)造器做了什么。

	final Thread thread; //為每個(gè)MyWorker

        MyWorker(){
            Thread td = threadFactory.newThread(this);
            td.setName(THREADPOOL_NAME+threadNumber.getAndIncrement());
            this.thread=td;
            this.thread.start();
            workers.add(this);
        }

MyWorker構(gòu)造器通過(guò)線(xiàn)程工廠對(duì)當(dāng)前對(duì)象生成Thread；
并設(shè)置線(xiàn)程名為：MyThread-Pool-自增線(xiàn)程編號(hào)；
然后調(diào)用線(xiàn)程的start方法啟動(dòng)線(xiàn)程；
最后存放在workers這個(gè)Set集合中，這樣就可以實(shí)現(xiàn)線(xiàn)程復(fù)用了。

手寫(xiě)線(xiàn)程池-默認(rèn)構(gòu)造器

	public MyThreadPoolExecutor(){
        this(5,new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory());
    }

默認(rèn)構(gòu)造器的賦初始值：
corePoolSize：5
waitingQueue：new ArrayBlockingQueue<>(10)，長(zhǎng)度為10的有限阻塞隊(duì)列
threadFactory：Executors.defaultThreadFactory()

手寫(xiě)線(xiàn)程池-execute方法

	public boolean execute(Runnable runnable)
    {
        return waitingQueue.offer(runnable);
    }

本質(zhì)上其實(shí)就是把Runnable（任務(wù)）放到waitingQueue中。

手寫(xiě)線(xiàn)程池-處理任務(wù)

	   @Override
        public void run() {
            //循環(huán)接收任務(wù)
                while (true)
                {
                    if((!isRunning&&waitingQueue.size()==0)||STOPNOW)
                    {
                        break;
                    }else {
                        Runnable runnable = waitingQueue.poll();
                        if(runnable!=null){
                            runnable.run();
                            System.out.println("task==>"+taskcount.incrementAndGet());
                        }
                    }
                }
        }

本質(zhì)上就是一個(gè)死循環(huán)接收任務(wù)，退出條件如下：

優(yōu)雅的退出。當(dāng)isRunning為false并且waitingQueue的隊(duì)列大小為0（也就是無(wú)任務(wù)了）
暴力退出。當(dāng)STOPNOW為true，則說(shuō)明調(diào)用了shutdownNow方法
else語(yǔ)句塊會(huì)不斷取任務(wù)，當(dāng)任務(wù)！=null時(shí)則調(diào)用run方法處理任務(wù)

手寫(xiě)線(xiàn)程池-優(yōu)雅關(guān)閉線(xiàn)程池

	public void shutdown()
    {
        this.isRunning=false;
    }

手寫(xiě)線(xiàn)程池-暴力關(guān)閉線(xiàn)程池

	public void shutdownNow()
    {
        this.STOPNOW=true;
    }

手寫(xiě)線(xiàn)程池-源代碼

手寫(xiě)線(xiàn)程池類(lèi)的源代碼

package com.springframework.concurrent;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 線(xiàn)程池類(lèi)
 * @author 游政杰
 */
public class MyThreadPoolExecutor {

    private final AtomicInteger taskcount=new AtomicInteger(0);//執(zhí)行任務(wù)次數(shù)
    private final AtomicInteger threadNumber=new AtomicInteger(0); //線(xiàn)程編號(hào)
    private volatile int corePoolSize; //核心線(xiàn)程數(shù)
    private final Set<MyThreadPoolExecutor.MyWorker> workers; //工作線(xiàn)程
    private final BlockingQueue<Runnable> waitingQueue; //等待隊(duì)列
    private final String THREADPOOL_NAME="MyThread-Pool-";//線(xiàn)程名稱(chēng)
    private volatile boolean isRunning=true; //是否運(yùn)行
    private volatile boolean STOPNOW=false; //是否立刻停止
    private final ThreadFactory threadFactory; //線(xiàn)程工廠

    public MyThreadPoolExecutor(){
        this(5,new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory());
    }

    public MyThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> waitingQueue,ThreadFactory threadFactory) {
        this.corePoolSize=corePoolSize;
        this.workers=new HashSet<>(corePoolSize);
        this.waitingQueue=waitingQueue;
        this.threadFactory=threadFactory;
        //線(xiàn)程預(yù)熱
        for (int i = 0; i < corePoolSize; i++) {
            new MyWorker();
        }
    }

    /**
     * MyWorker就是我們每一個(gè)線(xiàn)程對(duì)象
     */
    private final class MyWorker implements Runnable{

        final Thread thread; //為每個(gè)MyWorker

        MyWorker(){
            Thread td = threadFactory.newThread(this);
            td.setName(THREADPOOL_NAME+threadNumber.getAndIncrement());
            this.thread=td;
            this.thread.start();
            workers.add(this);
        }

        @Override
        public void run() {
            //循環(huán)接收任務(wù)
                while (true)
                {
                    //循環(huán)退出條件：
                    //1：當(dāng)isRunning為false并且waitingQueue的隊(duì)列大小為0（也就是無(wú)任務(wù)了），會(huì)優(yōu)雅的退出。
                    //2：當(dāng)STOPNOW為true，則說(shuō)明調(diào)用了shutdownNow方法進(jìn)行暴力退出。
                    if((!isRunning&&waitingQueue.size()==0)||STOPNOW)
                    {
                        break;
                    }else {
                        //不斷取任務(wù)，當(dāng)任務(wù)！=null時(shí)則調(diào)用run方法處理任務(wù)
                        Runnable runnable = waitingQueue.poll();
                        if(runnable!=null){
                            runnable.run();
                            System.out.println("task==>"+taskcount.incrementAndGet());
                        }
                    }
                }
        }
    }

    public boolean execute(Runnable runnable)
    {
        return waitingQueue.offer(runnable);
    }
    //優(yōu)雅的關(guān)閉
    public void shutdown()
    {
        this.isRunning=false;
    }
    //暴力關(guān)閉
    public void shutdownNow()
    {
        this.STOPNOW=true;
    }
}

測(cè)試使用手寫(xiě)線(xiàn)程池代碼

package com.springframework.test;

import com.springframework.concurrent.MyThreadPoolExecutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;

public class ThreadPoolTest {

  public static void main(String[] args) {

    
      MyThreadPoolExecutor myThreadPoolExecutor = new MyThreadPoolExecutor
              (5,new ArrayBlockingQueue<>(6), Executors.defaultThreadFactory());

      for(int i=0;i<10;i++){

          int finalI = i;
          myThreadPoolExecutor.execute(()->{
              System.out.println(Thread.currentThread().getName()+">>>>"+ finalI);
          });

      }

      myThreadPoolExecutor.shutdown();

//      myThreadPoolExecutor.shutdownNow();



  }
}