当前位置：首页 > news >正文

#简易线程池...实现原理

news 来源：原创 2025/6/4 16:38:53

代码实现（带详细注释）

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 简单线程池实现
 * 1. 使用阻塞队列管理待执行任务
 * 2. 固定数量的工作线程处理任务
 * 3. 支持优雅关闭
 */
public class SimpleThreadPool {
    // 任务队列：用于存储待执行的任务，采用线程安全的LinkedBlockingQueue实现
    private final BlockingQueue<Runnable> taskQueue;
    
    // 工作线程数组：用于执行任务的实际线程
    private final Worker[] workers;
    
    // 线程池状态标志：volatile保证多线程可见性
    private volatile boolean isStopped = false;

    /**
     * 构造函数：初始化线程池 
     * @param numThreads 线程池中工作线程的数量
     */
    public SimpleThreadPool(int numThreads) {
        // 创建无界任务队列（可根据需求改为有界队列）
        this.taskQueue = new LinkedBlockingQueue<>();
        
        // 初始化工作线程数组
        this.workers = new Worker[numThreads];

        // 创建并启动所有工作线程
        for (int i = 0; i < numThreads; i++) {
            workers[i] = new Worker();
            workers[i].start();  // 启动工作线程

            // !!!重点!!!! 是 SimpleThreadPool 的构造方法,在被创建后就会执行,开启的多个线程就start了

        }
    }

    /**
     * 提交任务到线程池
     * @param task 待执行的任务（Runnable接口实现）
     * @throws IllegalStateException 如果线程池已关闭
     */
    public void submit(Runnable task) {
        // 检查线程池状态
        if (!isStopped) {
            try {
                // 将任务放入队列（如果队列满会阻塞）
                taskQueue.put(task); 
            } catch (InterruptedException e) {
                // 恢复中断状态
                Thread.currentThread().interrupt();
            }
        } else {
            throw new IllegalStateException("ThreadPool is stopped");
        }
    }

    /**
     * 关闭线程池
     * 1. 设置停止标志
     * 2. 中断所有工作线程
     */
    public void shutdown() {
        // 设置停止标志
        isStopped = true;
        
        // 中断所有工作线程
        for (Worker worker : workers) {
            worker.interrupt(); 
        }
    }

    /**
     * 工作线程内部类
     * 负责从任务队列中获取并执行任务
     */
    private class Worker extends Thread {  
        @Override
        public void run() {
            // 只要线程池未关闭，就持续执行任务
            while (!isStopped) {
                try {
                    // 从队列获取任务（队列空时阻塞）
                    Runnable task = taskQueue.take(); 
                    
                    // 执行任务（同步执行）
                    task.run(); 
                    
                    /* !!!!重点!!!!
                    //我们都知道一个Runnable对象的run方法,在直接调用时,它是没有开启一个新的线程的
                    //在这个案例中,每一个Worker是一个线程, 其中执行runnable对象的run方法 , 
                    //   续上句: 这就相当于是 开了新的线程去执行Runnable对象的run方法 , 虽然这里对线程对象复用了,但原理是这样.
                    //线程对象 的任务是 从阻塞队列 取出Runnable对象, 并执行其run方法 
                    */
                    
                } catch (InterruptedException e) {
                    // 检查是否是因关闭线程池导致的中断
                    if (isStopped) {
                        break;  // 如果是关闭操作，则退出循环
                    }
                    // 其他中断情况继续循环
                }
            }
        }
    }

    /**
     * 测试主方法
     */
    public static void main(String[] args) {
        // 创建包含3个工作线程的线程池
        SimpleThreadPool threadPool = new SimpleThreadPool(3); 

        // 提交10个任务
        for (int i = 0; i < 10; i++) {
            final int taskNumber = i;
            threadPool.submit(() -> {
                System.out.println("开始执行任务 " + taskNumber + "，执行线程：" + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // 模拟任务执行耗时
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("完成任务 " + taskNumber);
            });
        }

        // 关闭线程池
        threadPool.shutdown(); 
    }
}

执行流程与原理详解

1. 初始化阶段

创建线程池实例：调用构造函数SimpleThreadPool(3)创建包含3个工作线程的线程池
初始化任务队列：创建LinkedBlockingQueue作为任务缓冲区
创建工作线程：
- 创建3个Worker线程实例
- 调用start()方法启动每个工作线程
工作线程启动：
- 每个Worker线程进入run()方法
- 立即调用taskQueue.take()尝试获取任务（此时队列为空，线程阻塞）