Java多线程实战-从零手搓一个简易线程池(二)线程池实现与拒绝策略接口定义

/**
 * @author Luckysj @刘仕杰
 * @description 自定义线程池对象
 * @create 2024/03/27 10:45:17
 */
@Slf4j
public class ThreadPool {
 
}

    /** 任务等待队列 */
    private WorkQueue<Runnable> workQueue;
    /** 正在运行的工作线程集合 */
    private final Set<Worker> workerSet = new HashSet<>();
    /** 核心线程数 */
    private int corePoolSize;
    /** 最大等待时间（也就是线程的最大空闲时间) */
    private Long keepAliveTime;
    /** 等待时间单位 */
    private TimeUnit timeUnit;
 
    public ThreadPool(WorkQueue<Runnable> workQueue, int corePoolSize, Long keepAliveTime, TimeUnit timeUnit) {
        this.workQueue = workQueue;
        this.corePoolSize = corePoolSize;
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = timeUnit;
    }

public void execute(Runnable task){
        synchronized(workerSet){
            //1 判断当前运行的工作线程数是否小于核心线程数
            if(workerSet.size() <  corePoolSize){
                // 2.1 创建工作线程
                Worker worker = new Worker(task);
                // 2.2 加入运行线程集合
                workerSet.add(worker);
                // 2.3 运行线程
                worker.start();
 
            }else{
                // 2.1 尝试将任务加入阻塞队列中等待(put方法会一直堵塞等待，后面会改进)
                workQueue.put(task);
            }
        }
    }

class Worker extends Thread{
        private Runnable task;
 
        public Worker(Runnable task) {
            this.task = task;
        }
 
        @Override
        public void run() {
            log.info("工作线程{}开始运行", Thread.currentThread());
 
            // 1。首先消费当前任务，消费完再去任务队列取，while循环实现线程复用
            while(task != null || (task = workQueue.poll(keepAliveTime, timeUnit)) != null){
                try {
                    task.run();
                }catch (Exception e){
                    throw new RuntimeException(e);
                }finally {
                    // 执行完后清除任务
                    task = null;
                }
            }
 
            // 2.跳出循环，说明取任务超过了最大等待时间，线程歇菜休息吧
            synchronized (workerSet){
                workerSet.remove(this);
            }
            log.info("线程{}超过最大空闲时间没有获取到任务，已被回收", Thread.currentThread());
        }
    }

public interface RejectPolicy<T> {
    void reject(BlockQueue<T> queue, T task);
}

// 尝试向队列添加任务，如果队列已满就触发拒绝策略
    public void tryPut(RejectPolicy<T> rejectPolicy, T task){
        lock.lock();
        try {
            if(deque.size() == size){
                // 队列满了就触发拒绝策略
                log.info("拒绝策略触发，当前任务：{}", task);
                rejectPolicy.reject(this, task);
            }else{
                // 队列没满就将任务加入队列
                log.debug("没有空闲线程，加入任务等待队列等待");
                deque.addLast(task);
                emptyCondition.signal();
            }
        }finally {
            lock.unlock();
        }
    }

public void execute(Runnable task){
        synchronized(workerSet){
            //1 判断当前运行的工作线程数是否小于核心线程数
            if(workerSet.size() <  corePoolSize){
                // 2.1 创建工作线程
                Worker worker = new Worker(task);
                // 2.2 加入运行线程集合
                workerSet.add(worker);
                // 2.3 运行线程
                worker.start();
 
            }else{
                // 2.1 尝试将任务加入阻塞队列中等待，如果加入失败，触发拒绝策略
                workQueue.tryPut(rejectPolicy, task);
 
            }
        }
    }

@Slf4j
public class MainTest {
    public static void main(String[] args) {
 
        ThreadPool threadPool = new ThreadPool(new WorkQueue<>(5), 2, 5L, TimeUnit.SECONDS,
                (queue, task) -> {
                    // 一直等
                    //queue.put(task);
                    // 调用者线程执行
                    //task.run();
                    // 直接抛出异常
                    // throw new RuntimeException("saa");
                    // 丢弃这个任务
                    log.debug("丢弃这个任务{}", task);
                });
 
       for (int i = 0; i < 4; i++) {
            threadPool.execute(() -> {
                System.out.println("执行任务------->当前执行线程为" + Thread.currentThread().toString());
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
        }
 
    }
}