Java中线程池自定义实现详解

什么是线程池

在Java中，每次创建线程都需要为该线程分配独立的资源，包括CPU利用时间、栈内存等，这些资源的分配和回收都需要时间开销。当并发任务数量较大时，频繁地创建线程会导致系统负担过重，极有可能会出现OOM等问题。为了解决这个问题，Java提供了线程池，它可以在系统初始化时创建一定数量的线程，并将这些线程保存在池中，执行任务时直接从池中取出线程，并在任务执行完成后归还线程。这种方式可以减少线程创建和销毁的次数，从而提高系统的性能、可靠性和稳定性。

Java自带线程池的工作原理

在Java中，线程池的实现主要通过Executor框架实现。Executor框架定义了执行服务接口Executor、ExecutorService、ScheduledExecutorService和线程工厂接口ThreadFactory。实际上，所有线程池的实现都基于它们。其中，Executor是线程池的根接口，ExecutorService和ScheduledExecutorService是Executor的扩展，而ThreadFactory则是线程创建工厂。下面是一个简单的示例：

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 提交任务
        for (int i = 0; i < 10; i++) {
            executorService.submit(new Task());
        }
        // 关闭线程池
        executorService.shutdown();
    }

    static class Task implements Runnable {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " is executing task.");
        }
    }
}

自定义线程池的实现

Java中线程池自定义实现需要创建一个实现Executor接口的类，并重写其execute方法，该方法就是线程池执行任务的核心方法，其具体逻辑就是从线程池中获取线程执行任务。以下是一个简单的示例：

public class CustomThreadPool implements Executor {
    private final int poolSize;
    private final BlockingQueue<Runnable> queue;
    private final Thread[] threads;

    public CustomThreadPool(int poolSize, int queueSize) {
        this.poolSize = poolSize;
        this.queue = new ArrayBlockingQueue<>(queueSize);
        this.threads = new Thread[poolSize];

        // 启动线程
        for (int i = 0; i < poolSize; i++) {
            threads[i] = new Thread(new Worker(queue));
            threads[i].start();
        }
    }

    @Override
    public void execute(Runnable command) {
        try {
            queue.put(command);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    static class Worker implements Runnable {
        private final BlockingQueue<Runnable> queue;

        public Worker(BlockingQueue<Runnable> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    queue.take().run();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

以上代码中，我们创建了一个CustomThreadPool类，它实现了Executor接口，并实现了execute方法。在构造函数中，我们创建了一个大小为poolSize的线程数组threads和一个大小为queueSize的阻塞队列queue。对于queue中的任务，我们创建一个Worker线程从队列中取出并执行。而execute方法就是将任务加入阻塞队列中。

示例1：自定义线程池使用

public class CustomThreadPoolExample1 {
    public static void main(String[] args) {
        CustomThreadPool threadPool = new CustomThreadPool(5, 10);
        for (int i = 0; i < 10; i++) {
            threadPool.execute(new Task(i));
        }
    }

    static class Task implements Runnable {
        private final int taskId;

        public Task(int taskId) {
            this.taskId = taskId;
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " is executing task " + taskId);
        }
    }
}

以上代码中，我们创建了一个大小为5的线程池，并提交了10个任务。可以看到，任务分别由5个线程执行，并且线程执行任务的顺序并不是按照提交的顺序执行的。

示例2：自定义线程池并增加异常处理

public class CustomThreadPoolExample2 {
    public static void main(String[] args) {
        CustomThreadPool threadPool = new CustomThreadPool(5, 10);
        for (int i = 0; i < 10; i++) {
            threadPool.execute(new Task(i));
        }
    }

    static class Task implements Runnable {
        private final int taskId;

        public Task(int taskId) {
            this.taskId = taskId;
        }

        @Override
        public void run() {
            try {
                // do something
                System.out.println(Thread.currentThread().getName() + " is executing task " + taskId);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
    }
}

以上代码中，我们在任务的run方法中增加了异常处理，这样就可以在任务执行时发现问题，并进行相应的处理。