Java手写线程池-第一代(原创)

举报
摸鱼打酱油 发表于 2022/04/01 22:27:00 2022/04/01
【摘要】 个人简介作者是一个来自河源的大三在校生,以下笔记都是作者自学之路的一些浅薄经验,如有错误请指正,将来会不断的完善笔记,帮助更多的Java爱好者入门。@[toc] Java手写线程池(第一代)经常使用线程池,故今天突发奇想,手写一个线程池,会有很多不足,请多多宽容。因为这也是第一代的版本,后续会更完善。 手写线程池-定义参数 private final AtomicInteger taskc...

个人简介

作者是一个来自河源的大三在校生,以下笔记都是作者自学之路的一些浅薄经验,如有错误请指正,将来会不断的完善笔记,帮助更多的Java爱好者入门。

@[toc]

Java手写线程池(第一代)

  • 经常使用线程池,故今天突发奇想,手写一个线程池,会有很多不足,请多多宽容。因为这也是第一代的版本,后续会更完善。

手写线程池-定义参数

	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:执行任务次数
threadNumber:线程编号,从0开始依次递增。
corePoolSize:核心线程数
workers:工作线程
waitingQueue:等待队列
THREADPOOL_NAME:线程名称
isRunning:是否运行
STOPNOW:是否立刻停止
threadFactory:线程工厂

手写线程池-构造器

    public MyThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> waitingQueue,ThreadFactory threadFactory) {
        this.corePoolSize=corePoolSize;
        this.workers=new HashSet<>(corePoolSize);
        this.waitingQueue=waitingQueue;
        this.threadFactory=threadFactory;
        //线程预热
        for (int i = 0; i < corePoolSize; i++) {
            new MyWorker();
        }
    }
  • 该构造器作用:
  • 1:对参数进行赋值。
  • 2:线程预热。根据corePoolSize的大小来调用MyWorker的构造器。我们可以看看MyWorker构造器做了什么。
	final Thread thread; //为每个MyWorker

        MyWorker(){
            Thread td = threadFactory.newThread(this);
            td.setName(THREADPOOL_NAME+threadNumber.getAndIncrement());
            this.thread=td;
            this.thread.start();
            workers.add(this);
        }
  • MyWorker构造器通过线程工厂对当前对象生成Thread
  • 设置线程名为:MyThread-Pool-自增线程编号;
  • 然后调用线程的start方法启动线程;
  • 最后存放在workers这个Set集合中,这样就可以实现线程复用了。

手写线程池-默认构造器

	public MyThreadPoolExecutor(){
        this(5,new ArrayBlockingQueue<>(10), Executors.defaultThreadFactory());
    }
  • 默认构造器的赋初始值
  • corePoolSize:5
  • waitingQueue:new ArrayBlockingQueue<>(10),长度为10的有限阻塞队列
  • threadFactory:Executors.defaultThreadFactory()

手写线程池-execute方法

	public boolean execute(Runnable runnable)
    {
        return waitingQueue.offer(runnable);
    }
  • 本质上其实就是把Runnable(任务)放到waitingQueue中

手写线程池-处理任务

	   @Override
        public void run() {
            //循环接收任务
                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());
                        }
                    }
                }
        }
  • 本质上就是一个死循环接收任务,退出条件如下:
  • 1:优雅的退出。当isRunning为false并且waitingQueue的队列大小为0(也就是无任务了)
  • 2:暴力退出。当STOPNOW为true,则说明调用了shutdownNow方法
  • else语句块会不断取任务,当任务!=null时则调用run方法处理任务

手写线程池-优雅关闭线程池

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

手写线程池-暴力关闭线程池

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

手写线程池-源代码

手写线程池类的源代码

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;

/**
 * 线程池类
 * @author 游政杰
 */
public class MyThreadPoolExecutor {

    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; //线程工厂

    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;
        //线程预热
        for (int i = 0; i < corePoolSize; i++) {
            new MyWorker();
        }
    }

    /**
     * MyWorker就是我们每一个线程对象
     */
    private final class MyWorker implements Runnable{

        final Thread thread; //为每个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() {
            //循环接收任务
                while (true)
                {
                    //循环退出条件:
                    //1:当isRunning为false并且waitingQueue的队列大小为0(也就是无任务了),会优雅的退出。
                    //2:当STOPNOW为true,则说明调用了shutdownNow方法进行暴力退出。
                    if((!isRunning&&waitingQueue.size()==0)||STOPNOW)
                    {
                        break;
                    }else {
                        //不断取任务,当任务!=null时则调用run方法处理任务
                        Runnable runnable = waitingQueue.poll();
                        if(runnable!=null){
                            runnable.run();
                            System.out.println("task==>"+taskcount.incrementAndGet());
                        }
                    }
                }
        }
    }

    public boolean execute(Runnable runnable)
    {
        return waitingQueue.offer(runnable);
    }
    //优雅的关闭
    public void shutdown()
    {
        this.isRunning=false;
    }
    //暴力关闭
    public void shutdownNow()
    {
        this.STOPNOW=true;
    }
}

测试使用手写线程池代码

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();



  }
}

问题:为什么自定义线程池的execute执行的任务有时会变少?

  • 那是因为waitingQueue满了放不下任务了,导致任务被丢弃,相当于DiscardPolicy拒绝策略
    • 解决办法有:
    • 1:设置最大线程数,自动对线程池扩容。
    • 2:调大waitingQueue的容量capacity

最后:因为这是我手写的线程池的初代版本,基本实现线程池的复用功能,然而还有很多未完善,将来会多出几篇完善后的文章,对目前手写的线程池进行升级。

后续还会继续出关于作者手写Spring框架,手写Tomcat等等框架的博文!!!!!

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。