Java多线程实战-从零手搓一个简易线程池(三)线程工厂，核心线程与非核心线程逻辑实现

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                // 线程执行任务流程，省流
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

/**
 * @author Luckysj @刘仕杰
 * @description 线程工厂接口
 * @create 2024/03/28 20:40:18
 */
public interface ThreadFactory {
    /**
    * @description
    * @param 
    * @return 创建的线程对象
    * @date 2024/03/28 21:01:35
    */
    Thread newThread(Runnable r);
}

/**
 * @author Luckysj @刘仕杰
 * @description 默认线程工厂，我们这里仿照源码写法，为每个线程分配线程组(默认会自动分配)，并为每个线程组
 * @create 2024/03/28 21:27:10
 */
public class DefaultThreadFactory implements ThreadFactory{
    /** 原子序号类，我们可以通过该类为线程工厂来获取一个随机序号，主要是为了区分不同线程池实例*/
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    /** 线程组，每个线程都需要属于一个线程组(平常使用未指定线程组会默认分配)*/
    private final ThreadGroup group;
    /** 原子序号类，我们可以通过该类为每个线程来获取一个随机序号*/
    private static final AtomicInteger threadNumber = new AtomicInteger(1);
    /** 线程名前缀，以便于在日志、监控等场景下识别和管理线程。*/
    private final String namePrefix;
 
    public DefaultThreadFactory() {
        // 获取管理安全策略的类，通过这个类我们可以获取对应名称的线程组，SecurityManager 和 group 的存在是为了更好地控制线程的安全性和权限
        SecurityManager s = System.getSecurityManager();
        // 存在 SecurityManager实例,则通过 s.getThreadGroup() 获取一个受限制的线程组。
        // 如果不存在 SecurityManager 实例,则使用当前线程所在的线程组 Thread.currentThread().getThreadGroup()。
        this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        // 生成前缀
        this.namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
    }
 
    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
        // 将线程设置为用户线程
        if(thread.isDaemon()){
            thread.setDaemon(false);
        }
        // 为线程设置默认优先级
        if(thread.getPriority() != Thread.NORM_PRIORITY){
            thread.setPriority(Thread.NORM_PRIORITY);
        }
        return thread;
    }
}

    class Worker implements Runnable{
        private Runnable firstTask;
 
        private Thread thread;
 
        public Worker(Runnable task) {
 
            this.firstTask = task;
            this.thread = threadFactory.newThread(this);
        }
        
        // 省略
    }

    /**
    * @description 从等待队列中获取任务
    * @return Runnable 待执行的任务，没有获取到会返回null
    * @date 2024/04/02 10:46:37
    */
public Runnable getTask(){
        //我们使用一个变量来记录上次循环获取任务是否超时
        boolean preIsTimeOut = false;
        // 内部使用一个while循环，线程会一直尝试获取任务，直到成功获取到任务或者满足退出条件
        while(true){
            // 获取线程池当前线程数量
            int wc = threadTotalNums.get();
            // 1.是否要进行核心线程回收操作，当allowCoreThreadTimeOut为true，或者当前线程池数大于核心线程数时，我们需要进行回收判断
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
            // 2.根据情况动态调整线程数，以下情况需要直接返回null(返回null就会回收线程):
            // (1)当前线程大于最大线程数(就是超过规定大小了)，且任务队列为空且存在工作线程
            // (2)timed为true，上次任务超时了(preIsTimeOut = true),且任务队列为空且存在工作
            if ( (wc > maximumPoolSize || (timed && preIsTimeOut)) && (wc > 1 || workQueue.isEmpty()) ) {
                return null;
            }
 
            // 3.根据timed这个条件来选择是超时堵塞
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            // 获取任务超时了，将preIsTimeOut设为true，下次可以执行回收
            preIsTimeOut = true;
        }
    }

 @Override
        public void run() {
            log.info("工作线程====》工作线程{}开始运行", Thread.currentThread());
 
            // 1。首先消费当前任务，消费完再去任务队列取，while循环实现线程复用
            while(firstTask != null || (firstTask = getTask()) != null){
                try {
                    firstTask.run();
                }catch (Exception e){
                    throw new RuntimeException(e);
                }finally {
                    // 执行完后清除任务
                    firstTask = null;
                }
            }
 
            // 2.跳出循环，说明取任务超过了最大等待时间，线程歇菜休息吧
            synchronized (workerSet){
                workerSet.remove(this);
                threadTotalNums.decrementAndGet(); //计数扣减
            }
            log.info("工作线程====》线程{}已被回收，当前线程数:{}", Thread.currentThread(), threadTotalNums.get());
 
        }

/**
    * @description 添加工作线程
    * @param firstTask 线程第一次执行的任务
    * @param isCore 是否为核心线程
    * @return Boolean 线程是否添加成功
    * @date 2024/04/02 10:42:43
    */
    public Boolean addWorker(Runnable firstTask, Boolean isCore){
        if(firstTask == null) {
            throw new NullPointerException();
        }
        // TODO 1.我们在添加线程时，首先可以进行一些与线程池生命周期相关的校验，比如在一些状态下，不允许再添加任务
 
        // 2.根据当前线程池和isCore条件判断是否需要创建
        int wc = threadTotalNums.get();
        if (wc >= (isCore ? corePoolSize : maximumPoolSize))
            return false;
        // 3.创建线程，并添加到线程集合中
        Worker worker = new Worker(firstTask);
        Thread t = worker.thread;
        if(t != null){
            synchronized (workerSet){
                workerSet.add(worker);
                threadTotalNums.getAndIncrement();
            }
            t.start();
            return true;
        }
        return false;
    }

public void execute(Runnable task){
        if(task == null){
            throw new NullPointerException("传递的Runnable任务为Null");
        }
        // 1.如果当前线程数小于核心线程,直接创建线程去运行
        if(threadTotalNums.get() < corePoolSize){
            if(addWorker(task, true)) return;
        }
 
        // 2.线程数大于核心线程，我们就将任务加入等待队列
        if(workQueue.offer(task)){
            return;
        }
        // 3.队列满了，尝试创建非核心线程，如果失败就触发拒绝策略
        else if(!addWorker(task, false)){
            reject(task);
        }
 
    }

@Slf4j
public class MainTest {
    public static void main(String[] args) {
 
        ThreadPool threadPool = new ThreadPool(new WorkQueue<>(5), 2, 5,5L, TimeUnit.SECONDS,
                (queue, task) -> {
                    log.info("拒绝策略====》拒绝策略触发，直接丢弃当前任务");
                }, new DefaultThreadFactory());
        threadPool.setAllowCoreThreadTimeOut(false); //不回收核心线程
        for (int i = 0; i < 15; i++) {
            threadPool.execute(() -> {
                System.out.println("执行任务------->当前执行线程为" + Thread.currentThread().toString());
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
        }
 
        // ExecutorService executorService = Executors.newFixedThreadPool(2);
 
    }
}

threadPool.setAllowCoreThreadTimeOut(true); //回收核心线程