手撕线程池 ThreadPool

为了更加方便理解线程池

我们都知道线程池和任务就好比生产者消费者之间的关系也就是如图！

先分析：我们可以将线程池中的线程看作消费者，可以将main(只是方便测试)看作生产者

总体思路：
实现一个阻塞队列，因为阻塞，阻塞队列是连接他们的桥梁，我们线程池中的线程去队列中获取任务执行，而我们main，提供任务到阻塞队列中供线程池执行！其次我们需要实现ThreadPool，我们可以在主线程当作消费者！

第一步：分析阻塞队列

我们阻塞队列当中应当包含一个任务队列，来记录我们待执行的任务
还应该给任务队列的对头元素加锁，这里使用ReentrantLock，加锁的目的，避免我们线程池中多个线程同时或许一个任务，保证线程安全；
同样我们的任务队列需要一个容量capacity ；
还需要两个条件变量Condition，（1）如果任务队列为空，线程池中的线程去休息，（2）如果任务队列满了，超出容量的任务需要去等待！

第二步：分析线程池

我们的线程池需要关联阻塞队列
我们的线程池需要保存多个线程类，可以用HashSet ；
我们的线程池因该具有核心线程数 coreSize ；
而且我们的线程池需要设置任务的超时时间，线程池中的线程会等待任务队列当中的任务，一旦超时线程池中的线程就不会再等待！
在为时间设置一个时间单位！
还需要一个成员内部类wordk，封装一下我们的Thread

第三步：将线程池和阻塞队列中的方法实现

阻塞队列中提供的的方法

线程池中的线程线程（消费者）获取当前队列中的任务的方法
生产者向任务队列当中提供任务
获取任务队列的大长度

线程池中的方法

首当其冲的就是要有一个执行任务的方法 excute(),（用于接收任务）
其次就是我们线程池中线程执行的run方法，执行我们的具体任务!（拿到任务，用于具体执行）

到这，我们手写线程池的东西都理清了代码如下

package com.pool;import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;/*
*   手写线程池ThreadPool
*/
public class TestPool {public static void main(String[] args) {//创建线程池对象ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 10);//模拟提供5个任务放给线程池，然而我们线程池设置2个核心线程！for (int i = 0; i < 5; i++) {int j = i ;threadPool.execute(()->{System.out.println("任务："+ j);});}}}/*
*   线程池
*/
class ThreadPool{//任务队列private BlockQueue<Runnable> taskQueue ;  //可以讲任务抽象为Runnable(可运行的！)//县线程集合private HashSet<Worker> workers = new HashSet() ;//核心线程数private int coreSize ;//获取任务的超时时间private long timeout ;//时间单位private TimeUnit timeUnit ;public void execute (Runnable task){//加锁保障workers的线程安全synchronized (workers){//如果我们的任务数 < coreSize , 直接创建空闲线程,执行！if (workers.size() < coreSize){Worker worker = new Worker(task);System.out.println("新增worker :" + worker);worker.start();  //线程执行执行任务workers.add(worker) ;} else {     //如果我们的任务数 > coreSize , 直接加入任务队列taskQueue.put(task);System.out.println("加入任务队列 :" + task);}}}public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockQueue<>(queueCapacity);}//将线程池中的线程包装为一个work类class Worker extends Thread{private Runnable task ;public Worker(Runnable task) {this.task = task;}@Override   //执行任务public void run() {//task不为空，直接执行任务//当task不为空任务队列有任务执行，队列中的任务// while(task != null || (task = taskQueue.take()) != null)  //死等，线程池中的线程一直等待获取任务队列中的任务while(task != null || (task = taskQueue.poll(timeout,timeUnit)) != null){  //等一会儿任务队列没有就不等了,停止线程，下次啥时候来，啥时候再新建try{System.out.println("正在运行任务：" + task);task.run();}catch (Exception e){e.printStackTrace();}finally {task = null ;}}synchronized (workers){System.out.println("线程执行任务结束，移处线程池 :" + this);workers.remove(this) ;}}}
}/*
*   阻塞队列
*/
class BlockQueue<T>{//1、任务队列private Deque<T> queue = new ArrayDeque<>() ;//2、锁 : 用于锁住对头元素，防止其被多个线程同时获取！private ReentrantLock  lock = new ReentrantLock() ;//3、生产者条件变量 (1) 任务队列满了，任务（生产者）就会进入WaitSetprivate Condition fullWaitSet = lock.newCondition() ;//4、消费者条件变量 (2) 任务队列为空，线程池中的线程（消费者）进WaitSetprivate Condition emptyWaitSet = lock.newCondition() ;//5、容量private int capacity ;public BlockQueue(int capacity) {this.capacity = capacity;}//增强我们线程获取当前队列中的任务的方法，如果队列为空，进入休息室等待(带时限的，不会死等)public T poll(long timeout , TimeUnit unit){lock.lock();try{long nanos = unit.toNanos(timeout) ;             //超时时间统一转换为纳秒while(queue.isEmpty()){        //队列为空，线程池中的线程去等待try {if(nanos <= 0) return null ;  //如果nanos <= 0 说明超时限了，直接唤醒nanos = emptyWaitSet.awaitNanos(nanos);  //如果被提前唤醒，返回nanos的剩余时间} catch (InterruptedException e) {e.printStackTrace();}}T task = queue.removeFirst();//队列不为空，获取并且移除队头任务fullWaitSet.signal(); //有空闲线程，唤醒正在休息的任务return task ; //返回取到的任务}finally {lock.unlock();}}//线程（消费者）获取任务队列当中的任务public T take(){lock.lock();try{while(queue.isEmpty()){        //队列为空，线程池中的线程去等待try {emptyWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}T task = queue.removeFirst();//队列不为空，获取并且移除队头任务fullWaitSet.signal(); //有空闲线程，唤醒正在休息的任务return task ;}finally {lock.unlock();}}//生产者向任务队列当中提供任务public void put(T element){lock.lock();try{while(queue.size() == capacity){    //任务队列已经满了，新任务去等待try {fullWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}queue.addLast(element);  //任务队列不满，添加任务到末尾emptyWaitSet.signal(); //队列中添加一个任务，唤醒一个线程池中正在休息的线程}finally {lock.unlock();}}//获取任务队列的大小public int size(){lock.lock();try{return queue.size();  //返回任务队列长度即可}finally {}}
}

测试结果：

发现与预期结果一致！

手撕线程池 ThreadPool相关推荐

C++ 中的多线程的使用和线程池建设。150行代码，手写线程池
C++ 11 引入了 std::thread 标准库,方便了多线程相关的开发工作. 说到多线程开发,可不仅仅是创建一个新线程就好了,不可避免的要涉及到线程同步的问题. 而保证线程同步,实现线程安全,就 ...
C#.Net使用线程池(ThreadPool)与专用线程(Thread)
线程池(ThreadPool)使用起来很简单,但它有一些限制: 1. 线程池中所有线程都是后台线程,如果进程的所有前台线程都结束了,所有的后台线程就会停止.不能把入池的线程改为前台线程. 2. 不能 ...
C# 线程池ThreadPool
什么是线程池?为什么要用线程池?怎么用线程池? 1. 什么是线程池? .NET Framework的ThreadPool类提供一个线程池,该线程池可用于执行任务.发送工作项.处理异步 I/O.代表其他 ...
【线程池】自行准备linux环境，带你手写线程池，只需仅仅150行代码|内存池|API|连接池|应用协议丨C/C++Linux服务器开发
[线程池]自行准备linux环境,带你手写线程池,只需仅仅150行代码视频讲解如下,点击观看: [线程池]自行准备linux环境,带你手写线程池,只需仅仅150行代码|内存池|API|连接池|应用协 ...
【线程池】自行准备linux环境，带你手写线程池，只需仅仅150行代码
[线程池]自行准备linux环境,带你手写线程池,只需仅仅150行代码视频讲解如下,点击观看: [线程池]自行准备linux环境,带你手写线程池,只需仅仅150行代码|内存池|API|连接池|应用协 ...
Java手写线程池-第一代（原创）
个人简介作者是一个来自河源的大三在校生,以下笔记都是作者自学之路的一些浅薄经验,如有错误请指正,将来会不断的完善笔记,帮助更多的Java爱好者入门. 文章目录个人简介 Java手写线程池(第一代) ...
threadpool的数量_多线程之旅(11)_如何限制系统线程池ThreadPool的最大最小并发数量_SetMaxThreads/SetMinThreads用法...
ThreadPool有两个设置线程池并发数量的方法,分别是: ThreadPool.SetMinThreads(int workerThreads, int completionPortThreads ...
线程池(ThreadPool)
线程池ThreadPool 1.线程池简介 2.线程池参数说明(重点) 2.1 常用参数说明 2.2 拒绝策略(重点) 3.线程池底层工作原理 4.入门案例 1.线程池简介线程池(英语:thre ...
Java线程池ThreadPool详解
Java线程池ThreadPool详解 1. 线程池概述 1.1 线程池简介 1.2 线程池特点 1.3 线程池解决问题 2. 线程池原理分析 2.1 线程池总体设计 2.6 线程池流转状态 2.2 ...

手撕线程池 ThreadPool

第一步：分析阻塞队列

第二步：分析线程池

第三步：将线程池和阻塞队列中的方法实现

手撕线程池 ThreadPool相关推荐

最新文章

热门文章

手撕线程池 ThreadPool

第一步：分析阻塞队列

第二步： 分析线程池

第三步：将线程池和阻塞队列中的方法实现

手撕线程池 ThreadPool相关推荐

最新文章

热门文章

第二步：分析线程池