Java手写线程池-第一代(原创)
【摘要】 个人简介作者是一个来自河源的大三在校生,以下笔记都是作者自学之路的一些浅薄经验,如有错误请指正,将来会不断的完善笔记,帮助更多的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)