iis服务器助手广告
返回顶部
首页 > 资讯 > 后端开发 > Python >学生视角手把手带你写Java 线程池初版
  • 642
分享到

学生视角手把手带你写Java 线程池初版

2024-04-02 19:04:59 642人浏览 薄情痞子

Python 官方文档:入门教程 => 点击学习

摘要

目录Java手写线程池(第一代)手写线程池-定义参数手写线程池-构造器手写线程池-默认构造器手写线程池-execute方法手写线程池-处理任务手写线程池-优雅关闭线程池手写线程池-暴

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

本质上就是一个死循环接收任务,退出条件如下:

  • 优雅的退出。当isRunning为false并且waitingQueue的队列大小为0(也就是无任务了)
  • 暴力退出。当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;


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

    
    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等等框架的博文!!!!!

到此这篇关于学生视角手把手带你写Java 线程池的文章就介绍到这了,更多相关Java 线程池内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

--结束END--

本文标题: 学生视角手把手带你写Java 线程池初版

本文链接: https://www.lsjlt.com/news/143247.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑,方便收藏和打印~

下载Word文档
猜你喜欢
软考高级职称资格查询
编程网,编程工程师的家园,是目前国内优秀的开源技术社区之一,形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容,为IT开发者提供了一个发现、使用、并交流开源技术的平台。
  • 官方手机版

  • 微信公众号

  • 商务合作