自定义线程池

文章目录

- 自定义线程池
- - 自定义线程池

线程池的概述

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

线程池的七大核心参数：

corePoolSize: 核心线程数目 - 池中会保留的最多线程数。
maximumPoolSize: 最大线程数目 - 核心线程+救急线程的最大数目。
keepAliveTime: 生存时间 - 救急线程的生存时间，生存时间内没有新任务，此线程资源会释放。
unit: 时间单位 - 救急线程的生存时间单位，如秒、毫秒等。
workQueue: - 当没有空闲核心线程时，新来任务会加入到此队列排队，队列满会创建救急线程执行任务。
threadFactory: 线程工厂 - 可以定制线程对象的创建，例如设置线程名字、是否是守护线程等。
handler: 拒绝策略 - 当所有线程都在繁忙，workQueue 也放满时，会触发拒绝策略。

（1）抛异常 java.util.concurrent.ThreadPoolExecutor.AbortPolicy。

（2）由调用者执行任务 java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy。

（3）丢弃任务 java.util.concurrent.ThreadPoolExecutor.DiscardPolicy。

（4）丢弃最早排队任务 java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy。

在这里插入图片描述

线程池的基本工作方式：

提交任务到线程池，线程池拿到任务后判断当前的线程池的线程数是否达到了最大线程数，如果没有达到当核心线程都没有空闲时就会创建救急线程来执行该任务，如果达到了最大线程线程数当前任务在等待队列没满的情况下就会加到任务的阻塞队列等待，队列满了就会执行对应的拒绝策略。

jdk提供的现场池：
- 固定大小线程池
- 定时任务线程池
- 单线程线程池
- 带缓冲的线程池

自定义线程池

等待队列

package com.qiumin;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author qiumin
 * @classname BlockingQueue
 * @Description love code
 * @date 2022-06-05 14:50
 */
public class BlockingQueue<T> {

    /**
     * 任务队列
     * */
    private final Deque<T> queue = new ArrayDeque<>();

    /**
     * 锁
     * */
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * 生产者条件变量，队列满时等待
     * */
    private final Condition fullWaitSet = lock.newCondition();

    /**
     * 消费者条件变量，队列空时等待
     * */
    private final Condition emptyWaitSet = lock.newCondition();

    /**
     * 队列容量
     * */
    private int capacity;

    /**
     * 构造器
     * */
    public BlockingQueue(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 = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }


    /**
     *阻塞获取
     * */
    public T take(){
        lock.lock();
        try {
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞添加
     * */
    public void put(T element){
        lock.lock();
        try {
            while (queue.size()==capacity){
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("任务加入阻塞队列中===="+element);
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 带超时
     * 阻塞添加
     * */
    public Boolean offer(T element,Long timeout, TimeUnit unit){
        lock.lock();
        try {
            Long nanos = unit.toNanos(timeout);
            while (queue.size()==capacity){
                try {
                    if (nanos<=0){
                        return false;
                    }
                    //返回剩余时间
                   nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("任务加入阻塞队列中===="+element);
            queue.addLast(element);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 带有拒绝策略
     * 尝试阻塞添加
     * */
    public void tryPut(ThreadPool.RejectPolicy rejectPolicy,T element){
        lock.lock();
        try {
            if(queue.size()==capacity){
                rejectPolicy.reject(this,element);
            }else {
                System.out.println("任务加入阻塞队列中===="+element);
                queue.addLast(element);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }



    /**
     * 获取大小
     * */
    public int getSize(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

}

线程池

package com.qiumin;

import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author qiumin
 * @classname ThreadPool
 * @Description love code
 * @date 2022-06-05 15:20
 */
public class ThreadPool {

    /**
     * 任务队列
     * */
    private BlockingQueue<Runnable> taskQueue;

    /**
     * 线程池
     * */
    private HashSet<Worker> workers = new HashSet<>();

    /**
     * 核心线程数
     * */
    private int coreSize;

    /**
     * 拒绝策略
     * */
    private RejectPolicy<Runnable> rejectPolicy;

    /**
     * 超时时间
     * */
    private long timeout;

    /**
     * 时间单位
     * */
    private TimeUnit timeUnit;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue=new BlockingQueue<>(queueCapacity);
        this.rejectPolicy=rejectPolicy;
    }

    interface RejectPolicy<T>{
        void reject(BlockingQueue<T> taskQueue,T task);
    }

    /**
     * 执行任务
     * */
    public void executor(Runnable task){
       synchronized (workers) {
           if(workers.size()<coreSize){
               Worker worker = new Worker(task);
               System.out.println("新增worker线程===="+worker+"----任务==="+task);
               workers.add(worker);
               worker.start();
           }else {
//               taskQueue.put(task);
               taskQueue.tryPut(rejectPolicy,task);
           }
       }
    }

    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {

            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("移除当前worker==="+this);
                workers.remove(this);
            }
        }

    }
}

测试

package com.qiumin;

import java.util.concurrent.TimeUnit;

/**
 * @author qiumin
 * @classname TestThreadPool
 * @Description love code
 * @date 2022-06-05 15:50
 */
public class TestThreadPool {
    public static void main(String[] args) {
        ThreadPool pool = new ThreadPool(1, 3000, TimeUnit.MILLISECONDS, 1,(taskQueue,task) -> {
            //死等
            //taskQueue.put(task);
            //超时等待
            taskQueue.offer(task,500L,TimeUnit.MILLISECONDS);
            //放弃任务
            //System.out.println("放弃任务");
            //抛出异常
            //throw new RuntimeException("异常");
            //主线程自己执行任务
            //task.run();
        });
        for (int i = 0; i < 3; i++) {
            int j=i;
            pool.executor(()->{
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(j);
            });

        }
    }
}

测试结果

在这里插入图片描述

qiumin

原文链接：https://blog.csdn.net/m0_57752520/article/details/125137143