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

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

目录java手写线程池（第一代）手写线程池-定义参数手写线程池-构造器手写线程池-默认构造器手http://写线程池-execute方法手写线程池-处理任务手写线程池-优雅关闭线程池手写线程池-暴力关闭线程池手写线程池-源代码问题

Java手写线程池（第一代）

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

手写线程池-定义参数

private final AtomicInteger taskcount=new AtomicInteger(0);

private final AtomicInteger threadNumber=new AtomicInteger(0);

private volatile int corePoolSize;

private final Set workers;

private final BlockingQueue 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 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：5waitingQueue：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;

/**

* 线程池类

* @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 workers; //工作线程

nvRkJwlzprivate final BlockingQueue 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 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;

nvRkJwlz 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

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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。