欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

图解线程池工作机制,手写线程池?

程序员文章站 2022-03-09 23:39:09
ThreadPoolExecutor构造函数的各个参数说明 public ThreadPoolExecutor(int corePoolSize,//线程池中核心线程数 int maximumPoolSize,//允许的最大线程数 long keepAliveTime,//线程空闲下来后,存活的时间 ......

ThreadPoolExecutor构造函数的各个参数说明

public ThreadPoolExecutor(int corePoolSize,//线程池中核心线程数
int maximumPoolSize,//允许的最大线程数
long keepAliveTime,//线程空闲下来后,存活的时间,这个参数只在> corePoolSize才有用
TimeUnit unit,//存活时间的单位值
BlockingQueue<Runnable> workQueue,//保存任务的阻塞队列
ThreadFactory threadFactory,//创建线程的工厂,给新建的线程赋予名字
RejectedExecutionHandler handler) //饱和策略,内置4种策略

AbortPolicy :直接抛出异常,默认;

CallerRunsPolicy:用调用者所在的线程来执行任务

DiscardOldestPolicy:丢弃阻塞队列里最老的任务,队列里最靠前的任务

DiscardPolicy :当前任务直接丢弃

并允许大佬实现自己的饱和策略,实现RejectedExecutionHandler接口即可。

 

jdk代码精选。

图解线程池工作机制,手写线程池? 

图解线程池工作机制,手写线程池?

 

 

 步骤一(对应图1号箭头,对应源码第一个if代码块):  线程池中线程数  小于corePoolSize  ,就会创建新线程。

 步骤二(对应图2号箭头,对应源码第二个if代码块):  线程池中线程数  等于corePoolSize  ,这个任务就会保存到BlockingQueue。

 步骤三(对应图3号箭头,对应源码第三个if代码块): BlockingQueue也满了, 线程池中线程数小于 maximumPoolSize时候就会再次创建新的线程。

 步骤四(对应图4号箭头,对应源码第三个if代码块): BlockingQueue满了, 线程池中线程数等于 maximumPoolSize时候就会执行饱和策略。

 

面试官:“谈谈你对线程池理解”,我:“默默的手写了一个线程池”,无形装逼最为致命。

 

/**
*
*类说明:自己线程池的实现
*/
public class MyThreadPool2 {
// 线程池中默认线程的个数为5
private static int WORK_NUM = 5;
// 队列默认任务个数为100
private static int TASK_COUNT = 100;

// 工作线程组
private WorkThread[] workThreads;

// 任务队列,作为一个缓冲
private final BlockingQueue<Runnable> taskQueue;
private final int worker_num;//用户在构造这个池,希望的启动的线程数

// 创建具有默认线程个数的线程池
public MyThreadPool2() {
this(WORK_NUM,TASK_COUNT);
}

// 创建线程池,worker_num为线程池中工作线程的个数
public MyThreadPool2(int worker_num,int taskCount) {
if (worker_num<=0) worker_num = WORK_NUM;
if(taskCount<=0) taskCount = TASK_COUNT;
this.worker_num = worker_num;
taskQueue = new ArrayBlockingQueue<>(taskCount);
workThreads = new WorkThread[worker_num];
for(int i=0;i<worker_num;i++) {
workThreads[i] = new WorkThread();
workThreads[i].start();
}
Runtime.getRuntime().availableProcessors();
}


// 执行任务,其实只是把任务加入任务队列,什么时候执行有线程池管理器决定
public void execute(Runnable task) {
try {
taskQueue.put(task);
} catch (InterruptedException e) {
e.printStackTrace();
}

}


// 销毁线程池,该方法保证在所有任务都完成的情况下才销毁所有线程,否则等待任务完成才销毁
public void destroy() {
// 工作线程停止工作,且置为null
System.out.println("ready close pool.....");
for(int i=0;i<worker_num;i++) {
workThreads[i].stopWorker();
workThreads[i] = null;//help gc
}
taskQueue.clear();// 清空任务队列
}

// 覆盖toString方法,返回线程池信息:工作线程个数和已完成任务个数
@Override
public String toString() {
return "WorkThread number:" + worker_num
+ " wait task number:" + taskQueue.size();
}

/**
* 内部类,工作线程
*/
private class WorkThread extends Thread{

@Override
public void run(){
Runnable r = null;
try {
while (!isInterrupted()) {
r = taskQueue.take();
if(r!=null) {
System.out.println(getId()+" ready exec :"+r);
r.run();
}
r = null;//help gc;
}
} catch (Exception e) {
// TODO: handle exception
}
}

public void stopWorker() {
interrupt();
}

}
}

 测试类:

 

public class TestMyThreadPool extends ThreadPoolExecutor{

public TestMyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
// TODO Auto-generated constructor stub
}

public static void main(String[] args) throws InterruptedException {
// 创建3个线程的线程池
MyThreadPool2 t = new MyThreadPool2(3,0);
t.execute(new MyTask("testA"));
t.execute(new MyTask("testB"));
t.execute(new MyTask("testC"));
t.execute(new MyTask("testD"));
t.execute(new MyTask("testE"));
System.out.println(t);
Thread.sleep(10000);
t.destroy();// 所有线程都执行完成才destory
System.out.println(t);
}

// 任务类
static class MyTask implements Runnable {

private String name;
private Random r = new Random();

public MyTask(String name) {
this.name = name;
}

public String getName() {
return name;
}

@Override
public void run() {// 执行任务
try {
Thread.sleep(r.nextInt(1000)+2000);
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getId()+" sleep InterruptedException:"
+Thread.currentThread().isInterrupted());
}
System.out.println("任务 " + name + " 完成");
}
}
}