线程池：高效管理并发任务的利器

news2024/10/21 5:35:14

线程池：高效管理并发任务的利器

什么是线程池？

线程池（Thread Pool）是Java并发编程中的一种设计模式，旨在通过重复利用线程资源，来提高程序执行效率。线程池的主要思想是提前创建一组可供使用的线程，这些线程在完成任务后并不会销毁，而是返回线程池，等待处理下一个任务。这种方式减少了频繁创建和销毁线程的开销。

线程池能解决的问题：

减少线程创建销毁的开销：频繁创建和销毁线程会影响性能，尤其是在高并发场景下。
控制最大并发量：线程池可以限制同时执行的线程数，避免因创建过多线程导致系统资源耗尽。
线程管理：线程池统一管理线程的生命周期，包括创建、执行、销毁等操作。

如何实现线程池？

在Java中，线程池的实现主要依赖于 java.util.concurrent 包中的 Executor 框架，其中最常用的实现类是 ThreadPoolExecutor。

1. 创建线程池

使用 Executors 工具类可以轻松创建常用的线程池，如固定线程池、缓存线程池、单线程池等。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个固定大小的线程池，线程池大小为3
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        
        // 提交任务到线程池
        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            threadPool.submit(() -> {
                System.out.println("Task " + taskId + " is running by thread " + Thread.currentThread().getName());
                try {
                    // 模拟任务耗时
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Task " + taskId + " completed.");
            });
        }
        
        // 关闭线程池
        threadPool.shutdown();
    }
}

2. 代码解释

创建线程池：Executors.newFixedThreadPool(3) 创建了一个包含3个线程的固定线程池。
提交任务：通过 threadPool.submit() 提交任务到线程池，任务被线程池中的线程执行。
模拟任务：每个任务通过 Thread.sleep(1000) 模拟耗时1秒。
关闭线程池：shutdown() 方法在所有任务执行完毕后关闭线程池。

3. 运行结果

Task 1 is running by thread pool-1-thread-1
Task 2 is running by thread pool-1-thread-2
Task 3 is running by thread pool-1-thread-3
Task 1 completed.
Task 4 is running by thread pool-1-thread-1
Task 2 completed.
Task 5 is running by thread pool-1-thread-2
Task 3 completed.
Task 4 completed.
Task 5 completed.

解释：最多3个任务可以同时执行，超过的任务需要等待线程池中的线程空闲后才能执行。

如何手动实现线程池？

你也可以通过自己实现一个简单的线程池来了解其原理。下面展示了一个基本的线程池实现。

import java.util.LinkedList;
import java.util.List;

class SimpleThreadPool {
    private final List<Worker> workers; // 存储线程池中的线程
    private final LinkedList<Runnable> taskQueue; // 任务队列
    private volatile boolean isShutdown = false; // 用于标记线程池是否关闭

    public SimpleThreadPool(int poolSize) {
        taskQueue = new LinkedList<>();
        workers = new LinkedList<>();

        for (int i = 0; i < poolSize; i++) {
            Worker worker = new Worker();  // 创建工作线程
            workers.add(worker);
            worker.start();  // 启动线程
        }
    }

    // 提交任务到任务队列
    public synchronized void submit(Runnable task) {
        if (!isShutdown) {
            taskQueue.add(task);
            notify();  // 唤醒等待的工作线程
        }
    }

    // 关闭线程池
    public synchronized void shutdown() {
        isShutdown = true;
        for (Worker worker : workers) {
            worker.interrupt();  // 中断所有工作线程
        }
    }

    private class Worker extends Thread {
        public void run() {
            while (true) {
                Runnable task;
                synchronized (SimpleThreadPool.this) {
                    while (taskQueue.isEmpty() && !isShutdown) {
                        try {
                            SimpleThreadPool.this.wait();  // 等待有新任务到来
                        } catch (InterruptedException e) {
                            return;  // 如果线程被中断，退出
                        }
                    }
                    task = taskQueue.poll();  // 从任务队列中取出任务
                }

                if (task != null) {
                    task.run();  // 执行任务
                }

                if (isShutdown && taskQueue.isEmpty()) {
                    return;  // 如果关闭并且没有任务了，退出线程
                }
            }
        }
    }
}

代码解释：

taskQueue：任务队列，用来存储待执行的任务。
Worker 类：工作线程从任务队列中取任务并执行，空闲时等待新的任务。
submit() 方法：用于提交任务到任务队列，同时通知工作线程执行。
shutdown() 方法：关闭线程池并中断所有工作线程。

使用示例：

public class CustomThreadPoolDemo {
    public static void main(String[] args) {
        // 创建线程池，大小为3
        SimpleThreadPool pool = new SimpleThreadPool(3);
        
        // 提交任务
        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            pool.submit(() -> {
                System.out.println("Task " + taskId + " is running by thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // 模拟任务耗时
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Task " + taskId + " completed.");
            });
        }
        
        // 关闭线程池
        pool.shutdown();
    }
}