MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java线程池的正确使用方式

2021-11-273.3k 阅读

线程池概述

在Java多线程编程中，线程池是一种重要的工具，它可以帮助我们管理和复用线程，从而提高应用程序的性能和资源利用率。线程池维护着一个线程队列，当有任务需要执行时，线程池会从队列中取出一个线程来执行任务，任务执行完毕后，线程并不会被销毁，而是返回到线程池中等待下一个任务。这样就避免了频繁创建和销毁线程带来的开销。

线程池的优势

  1. 提高性能：由于线程的创建和销毁开销较大，线程池通过复用已有的线程，减少了这种开销，从而提高了应用程序的响应速度和吞吐量。
  2. 资源控制：线程池可以限制线程的数量，避免因线程过多导致系统资源耗尽，从而提高系统的稳定性和可靠性。
  3. 便于管理：线程池提供了统一的接口来管理线程，例如提交任务、关闭线程池等操作，使得多线程编程更加简洁和易于维护。

Java线程池的实现原理

Java线程池的核心实现类是ThreadPoolExecutor，它实现了ExecutorService接口。ThreadPoolExecutor内部维护了几个重要的参数：

  1. 核心线程数（corePoolSize）：线程池中会一直存活的线程数量，即使这些线程处于空闲状态，也不会被销毁，除非设置了allowCoreThreadTimeOut为true。
  2. 最大线程数（maximumPoolSize）：线程池中允许的最大线程数量。当任务队列已满且核心线程都在忙碌时，线程池会创建新的线程，直到线程数量达到最大线程数。
  3. 任务队列（workQueue）：用于存储等待执行的任务。当核心线程都在忙碌时，新提交的任务会被放入任务队列中等待执行。
  4. 线程存活时间（keepAliveTime）：当线程数量超过核心线程数时，多余的空闲线程在等待新任务的时间超过该值后，会被销毁。
  5. 拒绝策略（RejectedExecutionHandler）：当任务队列已满且线程数量达到最大线程数时，新提交的任务会被拒绝，此时会执行拒绝策略。

Java线程池的创建方式

在Java中，我们可以通过ThreadPoolExecutor的构造函数来创建线程池，也可以使用Executors工具类提供的静态方法来创建不同类型的线程池。

使用ThreadPoolExecutor构造函数创建线程池

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

public class CustomThreadPoolExample {
    public static void main(String[] args) {
        // 创建任务队列
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5, // 核心线程数
                10, // 最大线程数
                10, // 线程存活时间
                TimeUnit.SECONDS,
                workQueue,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

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

在上述代码中，我们通过ThreadPoolExecutor的构造函数创建了一个线程池，核心线程数为5，最大线程数为10，线程存活时间为10秒，任务队列使用LinkedBlockingQueue，容量为10，并使用CallerRunsPolicy拒绝策略。然后我们提交了20个任务，观察线程池的执行情况。

使用Executors工具类创建线程池

  1. FixedThreadPool：创建一个固定大小的线程池，线程池中的线程数量始终保持不变。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolExample {
    public static void main(String[] args) {
        // 创建固定大小的线程池
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}
  1. CachedThreadPool：创建一个可缓存的线程池，如果线程池中的线程空闲时间超过60秒，就会被销毁。线程池的大小会根据任务数量自动调整。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolExample {
    public static void main(String[] args) {
        // 创建可缓存的线程池
        ExecutorService executor = Executors.newCachedThreadPool();

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}
  1. SingleThreadExecutor：创建一个单线程的线程池，线程池中只有一个线程，所有任务会按照提交的顺序依次执行。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        // 创建单线程的线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}
  1. ScheduledThreadPool：创建一个支持定时任务和周期性任务的线程池。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolExample {
    public static void main(String[] args) {
        // 创建支持定时任务的线程池
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

        // 提交延迟任务
        executor.schedule(() -> {
            System.out.println("Delayed task is running on thread " + Thread.currentThread().getName());
        }, 5, TimeUnit.SECONDS);

        // 提交周期性任务
        executor.scheduleAtFixedRate(() -> {
            System.out.println("Periodic task is running on thread " + Thread.currentThread().getName());
        }, 0, 2, TimeUnit.SECONDS);

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

线程池的任务提交与执行

在创建好线程池后，我们可以通过submit或execute方法来提交任务。execute方法用于提交不需要返回值的任务，而submit方法用于提交需要返回值的任务。

使用execute方法提交任务

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

public class ExecuteTaskExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 使用execute方法提交任务
        executor.execute(() -> {
            System.out.println("Task is running on thread " + Thread.currentThread().getName());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

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

使用submit方法提交任务

import java.util.concurrent.*;

public class SubmitTaskExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 使用submit方法提交任务并获取返回值
        Future<String> future = executor.submit(() -> {
            System.out.println("Task is running on thread " + Thread.currentThread().getName());
            Thread.sleep(1000);
            return "Task completed";
        });

        try {
            String result = future.get();
            System.out.println("Task result: " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

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

在上述代码中，我们通过submit方法提交了一个有返回值的任务，并通过Future对象获取任务的执行结果。

线程池的拒绝策略

当任务队列已满且线程数量达到最大线程数时，新提交的任务会被拒绝，此时会执行拒绝策略。Java提供了几种内置的拒绝策略：

  1. AbortPolicy：默认的拒绝策略，当任务被拒绝时，会抛出RejectedExecutionException异常。
  2. CallerRunsPolicy：当任务被拒绝时，会在调用execute或submit方法的线程中直接执行被拒绝的任务。
  3. DiscardPolicy：当任务被拒绝时，直接丢弃该任务，不做任何处理。
  4. DiscardOldestPolicy：当任务被拒绝时，会丢弃任务队列中最老的一个任务，然后尝试重新提交当前任务。

我们可以在创建线程池时通过构造函数来指定拒绝策略，例如：

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

public class RejectedExecutionHandlerExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                10,
                TimeUnit.SECONDS,
                workQueue,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

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

在上述代码中，我们使用了CallerRunsPolicy拒绝策略。

线程池的关闭

当我们不再需要使用线程池时，需要及时关闭线程池，以释放资源。线程池提供了两种关闭方式：shutdown和shutdownNow。

shutdown方法

shutdown方法会启动一个有序关闭过程，不再接受新的任务，但会继续执行已提交到任务队列中的任务。

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

public class ShutdownThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

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

shutdownNow方法

shutdownNow方法会尝试停止所有正在执行的任务，停止处理等待队列中的任务，并返回等待执行的任务列表。

import java.util.List;
import java.util.concurrent.*;

public class ShutdownNowThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 关闭线程池
        List<Runnable> tasks = executor.shutdownNow();
        System.out.println("Number of tasks not executed: " + tasks.size());
    }
}

线程池的监控与调优

在实际应用中，我们需要对线程池进行监控和调优，以确保其性能和稳定性。

线程池的监控

我们可以通过ThreadPoolExecutor提供的一些方法来监控线程池的状态，例如：

  1. getPoolSize：获取当前线程池中的线程数量。
  2. getActiveCount：获取当前正在执行任务的线程数量。
  3. getQueue：获取任务队列。
  4. getTaskCount：获取已提交到线程池的任务总数。
  5. getCompletedTaskCount：获取已完成的任务数量。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolMonitoringExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                10,
                TimeUnit.SECONDS,
                workQueue
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 监控线程池状态
        System.out.println("Pool size: " + executor.getPoolSize());
        System.out.println("Active count: " + executor.getActiveCount());
        System.out.println("Queue size: " + executor.getQueue().size());
        System.out.println("Task count: " + executor.getTaskCount());
        System.out.println("Completed task count: " + executor.getCompletedTaskCount());

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

线程池的调优

线程池的调优主要涉及到合理设置核心线程数、最大线程数、任务队列大小等参数。调优的关键在于根据应用程序的实际需求和运行环境进行测试和调整。

  1. 核心线程数的设置：核心线程数应该根据应用程序的CPU密集型或I/O密集型特性来设置。对于CPU密集型任务，核心线程数一般设置为CPU核心数 + 1；对于I/O密集型任务，核心线程数可以适当增加，例如设置为2 * CPU核心数。
  2. 最大线程数的设置：最大线程数应该根据系统资源和任务负载来设置，避免线程过多导致系统资源耗尽。
  3. 任务队列大小的设置：任务队列大小应该根据任务的到达速率和处理速率来设置，避免任务队列过长导致内存溢出。

线程池在实际项目中的应用场景

  1. Web服务器：在Web服务器中，线程池可以用于处理客户端请求，提高服务器的并发处理能力。
  2. 数据库操作：在进行大量数据库查询或更新操作时，可以使用线程池来提高操作效率。
  3. 文件处理：在处理大量文件时，线程池可以并行处理文件，加快处理速度。

总结

正确使用Java线程池可以显著提高应用程序的性能和资源利用率。通过合理设置线程池的参数、选择合适的拒绝策略、及时关闭线程池以及进行有效的监控和调优，我们可以充分发挥线程池的优势，打造高效稳定的多线程应用程序。在实际项目中，需要根据具体的业务场景和需求来选择合适的线程池类型和配置参数，以实现最佳的性能和用户体验。