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

Java Executors 的线程池创建策略

2021-05-062.7k 阅读

Java Executors 的线程池创建策略

线程池概述

在Java多线程编程中，线程池是一种非常重要的工具。线程池可以管理一组工作线程，通过复用这些线程来执行多个任务，从而避免了频繁创建和销毁线程带来的开销，提高了系统的性能和资源利用率。Java通过java.util.concurrent.Executors类提供了便捷的方式来创建不同类型的线程池，每个线程池创建策略都有其特定的适用场景。

Executors创建线程池的主要方法

  1. newFixedThreadPool(int nThreads)
    • 策略描述：创建一个固定大小的线程池，线程池中的线程数量始终保持为nThreads。当有新任务提交时，如果线程池中有空闲线程，则立即执行任务；如果没有空闲线程，任务会被放入阻塞队列中等待，直到有线程空闲。
    • 适用场景：适用于已知工作量且相对稳定的场景，比如服务器处理固定数量的客户端请求，或者执行一些批量处理任务，确保资源使用的稳定性。
    • 代码示例：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 5; i++) {
            int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
}

在上述代码中，创建了一个大小为3的固定线程池。提交5个任务，由于线程池只有3个线程，所以一开始3个任务会并行执行，另外2个任务会在队列中等待，当有线程完成任务释放后，等待的任务会被执行。

  1. newSingleThreadExecutor()
    • 策略描述：创建一个单线程的线程池，线程池中只有一个线程来执行任务。所有任务按照提交的顺序依次执行，前一个任务执行完毕后，才会执行下一个任务。
    • 适用场景：适用于需要顺序执行任务，且不希望有并发干扰的场景，比如数据库的某些单线程操作，确保数据的一致性。
    • 代码示例：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 5; i++) {
            int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
}

此代码创建了一个单线程线程池，5个任务会依次在这个单线程中执行，每个任务间隔2秒。

  1. newCachedThreadPool()
    • 策略描述：创建一个可缓存的线程池。如果线程池中有空闲线程，则复用空闲线程执行新任务；如果没有空闲线程，则创建新线程执行任务。当线程空闲时间超过60秒，线程会被回收。
    • 适用场景：适用于执行大量短时间任务的场景，例如Web服务器处理大量的HTTP请求，因为它能快速响应任务请求，同时在任务减少时自动回收线程资源。
    • 代码示例：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
}

在这个示例中，一开始可能会创建多个线程来并行执行任务，随着任务的完成，线程如果在60秒内没有新任务，就会被回收。

  1. newScheduledThreadPool(int corePoolSize)
    • 策略描述：创建一个大小固定的线程池，支持定时及周期性任务执行。corePoolSize指定了线程池中的核心线程数量，这些核心线程即使空闲也不会被回收。
    • 适用场景：适用于需要定时执行任务或者周期性执行任务的场景，比如定时备份数据库、定时清理缓存等。
    • 代码示例：
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 scheduledExecutorService = Executors.newScheduledThreadPool(2);
        scheduledExecutorService.schedule(() -> {
            System.out.println("Delayed task is running on thread " + Thread.currentThread().getName());
        }, 3, TimeUnit.SECONDS);

        scheduledExecutorService.scheduleAtFixedRate(() -> {
            System.out.println("Periodic task is running on thread " + Thread.currentThread().getName());
        }, 1, 2, TimeUnit.SECONDS);
    }
}

上述代码中，首先使用schedule方法延迟3秒执行一个任务，然后使用scheduleAtFixedRate方法以固定速率执行任务，初始延迟1秒，之后每2秒执行一次。

  1. newWorkStealingPool(int parallelism)
    • 策略描述：创建一个拥有多个任务队列的线程池，线程池中的线程会从其他线程的任务队列中窃取任务来执行。parallelism参数指定了线程池的并行级别，即线程池中线程的大致数量。如果不指定parallelism，则默认使用Runtime.getRuntime().availableProcessors()返回的值。
    • 适用场景：适用于执行大量可并行任务的场景，特别是那些任务执行时间较长且可以被分割为多个子任务的情况。例如，并行计算、数据挖掘等领域，通过工作窃取算法可以有效地提高系统的整体性能。
    • 代码示例：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class WorkStealingPoolExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newWorkStealingPool(4);
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        while (!executorService.isTerminated()) {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executorService.shutdown();
    }
}

在这个示例中，创建了一个并行级别为4的工作窃取线程池。提交10个任务，线程会从其他线程的任务队列中窃取任务执行，以提高整体执行效率。

线程池创建策略的本质分析

  1. 资源管理与复用
    • 无论是固定大小的线程池、单线程线程池还是可缓存线程池，其核心目的之一都是对线程资源进行有效的管理和复用。通过创建线程池，避免了在每次执行任务时都创建新线程的开销。线程的创建和销毁涉及到操作系统资源的分配和回收，这是相对昂贵的操作。以newFixedThreadPool为例，固定数量的线程在任务执行过程中被复用，当任务完成后，线程不会被销毁，而是等待下一个任务，大大减少了线程创建和销毁的频率。
    • 在newCachedThreadPool中，虽然线程在空闲60秒后会被回收，但在任务密集时，线程会被缓存复用，同样提高了资源的利用效率。
  2. 任务调度与执行顺序
    • 不同的线程池创建策略决定了任务的调度和执行顺序。newSingleThreadExecutor确保任务按照提交顺序依次执行，这是通过内部的单线程机制实现的。所有任务都被放入一个队列中，由唯一的线程依次从队列中取出任务并执行，这种机制保证了任务执行的顺序性，避免了并发执行可能带来的资源竞争和数据不一致问题。
    • 而newFixedThreadPool和newCachedThreadPool等多线程线程池，任务的执行顺序取决于线程池的调度策略。任务首先被提交到队列中，然后由线程池中的线程从队列中获取任务执行。在固定线程池newFixedThreadPool中，如果任务提交速度超过线程执行速度，任务会在队列中等待，直到有线程空闲。这种调度策略在一定程度上保证了任务的公平性，不会出现某个任务长时间得不到执行的情况。
  3. 应对不同任务负载
    • 不同的线程池创建策略适用于不同的任务负载情况。newFixedThreadPool适合处理相对稳定的任务负载，其固定的线程数量可以根据任务的预计数量和复杂度进行合理配置。例如，一个Web服务器可能知道它通常需要处理的并发请求数量，通过配置合适大小的固定线程池，可以有效地利用系统资源，避免线程过多导致的上下文切换开销和资源耗尽问题。
    • newCachedThreadPool则更适合处理突发的、短时间内大量的任务负载。当任务突然增加时，它可以迅速创建新线程来处理任务，而当任务减少时，又能自动回收空闲线程，释放资源。这种灵活性使得它在应对动态变化的任务负载时表现出色，例如处理高并发的HTTP请求。
    • newScheduledThreadPool主要应对定时和周期性任务负载，通过其内部的调度机制，能够准确地按照设定的时间间隔执行任务，满足诸如定时数据备份、系统监控等场景的需求。
  4. 工作窃取算法的本质
    • newWorkStealingPool所采用的工作窃取算法是一种高效的并行任务执行策略。在多线程环境下，每个线程都有自己的任务队列。当某个线程完成了自己队列中的任务后，它会从其他线程的任务队列的尾部窃取任务来执行。这种算法的核心思想是充分利用线程的空闲时间，提高整体系统的并行度和资源利用率。
    • 例如，在一个多核处理器系统中，不同的线程可能在不同的核心上执行任务。如果某个核心上的线程任务执行完毕，而其他核心上的线程任务队列中还有大量任务，那么空闲的线程就可以通过工作窃取算法从其他线程的队列中获取任务，从而避免了核心资源的浪费，提高了整个系统的性能。

线程池参数的深入理解

  1. 核心线程数（Core Pool Size）
    • 对于newFixedThreadPool，核心线程数就是创建线程池时指定的nThreads。这些核心线程在任务执行期间会一直存在，即使它们处于空闲状态，也不会被回收（除非设置了allowCoreThreadTimeOut为true）。核心线程数决定了线程池能够同时处理的任务数量的下限，在任务负载稳定的情况下，核心线程能够满足大部分任务的执行需求，避免频繁创建和销毁线程带来的开销。
    • 在newScheduledThreadPool中，corePoolSize指定了线程池中的核心线程数量，这些核心线程用于执行定时和周期性任务。由于定时任务的特殊性，需要保证一定数量的线程始终可用，以确保任务能够按时执行。
  2. 最大线程数（Maximum Pool Size）
    • 在newCachedThreadPool中，虽然没有明确设置最大线程数，但实际上它可以根据任务的需求动态创建新线程，理论上最大线程数可以达到Integer.MAX_VALUE。这是因为它的设计目的是为了快速响应大量短时间任务，在任务量突然增加时能够迅速创建足够的线程来处理。
    • 对于newFixedThreadPool，最大线程数等于核心线程数，即固定大小的线程池不会创建超过核心线程数的线程。这种设计使得线程池的资源使用相对稳定，不会因为任务量的波动而无限增加线程数量，避免了系统资源的过度消耗。
  3. 阻塞队列（Blocking Queue）
    • 线程池使用阻塞队列来存储等待执行的任务。不同类型的线程池使用的阻塞队列有所不同。newFixedThreadPool默认使用LinkedBlockingQueue，这是一个无界队列。无界队列意味着在任务提交速度超过线程执行速度时，任务会不断在队列中堆积，不会抛出异常。这种设计保证了任务不会因为队列满而丢失，但可能会导致内存占用不断增加，在极端情况下可能会引发内存溢出问题。
    • newCachedThreadPool默认使用SynchronousQueue，这是一个同步队列，它不存储任务，而是直接将任务交给线程执行。如果没有可用线程，任务会等待，直到有线程可用。这种队列适用于任务执行速度较快的场景，因为它不会缓存任务，减少了任务在队列中的等待时间。
  4. 线程存活时间（Keep - Alive Time）
    • 在newCachedThreadPool中，线程存活时间为60秒。当线程空闲时间超过这个时长，线程会被回收。这一机制确保了在任务量减少时，线程池能够自动释放空闲线程，避免资源浪费。线程存活时间的设置需要根据实际应用场景进行调整，如果任务执行频率波动较大，较短的存活时间可以更快地回收资源；如果任务执行频率相对稳定，较长的存活时间可以减少线程创建和销毁的开销。

实际应用中的选择与优化

  1. 根据任务特性选择线程池

    • 如果任务是CPU密集型的，即任务主要消耗CPU资源，例如复杂的数学计算、加密解密等，应该选择较小的线程池。对于newFixedThreadPool，核心线程数可以设置为Runtime.getRuntime().availableProcessors()，这样可以充分利用CPU核心，避免过多线程导致的上下文切换开销。例如，在一个数据挖掘算法中，对大量数据进行复杂的分析计算，使用固定大小且与CPU核心数匹配的线程池能够提高计算效率。
    • 如果任务是I/O密集型的，如文件读写、网络通信等，由于I/O操作通常会使线程处于等待状态，此时可以适当增加线程数量。对于newFixedThreadPool，核心线程数可以设置为Runtime.getRuntime().availableProcessors() * 2甚至更多，以充分利用等待I/O的时间执行其他任务。比如在一个网络爬虫程序中，线程需要等待网络响应，较多的线程可以提高整体的爬取效率。
    • 如果任务是混合型的，即既有CPU密集型操作又有I/O密集型操作，需要综合考虑任务的比例和特性来选择合适的线程池和线程数量。可以通过性能测试来确定最优的配置。

  2. 线程池的监控与调优

    • 在实际应用中，需要对线程池进行监控，了解线程池的运行状态，如线程数量、任务队列大小、任务执行时间等。可以通过ThreadPoolExecutor类提供的一些方法来获取这些信息，例如getPoolSize()获取当前线程池中的线程数量，getQueue().size()获取任务队列的大小等。
    • 根据监控数据进行调优。如果发现任务队列经常满，说明线程池处理任务的速度跟不上任务提交的速度，可能需要增加线程数量或者优化任务执行逻辑。如果发现线程空闲时间过长，说明线程池可能设置过大，可以适当减少线程数量，以提高资源利用率。
    • 另外，还可以考虑使用一些性能监控工具，如Java VisualVM、YourKit等，这些工具可以直观地展示线程池的运行状态，帮助开发者更好地进行调优。

  3. 异常处理与健壮性设计

    • 在使用线程池时，需要考虑任务执行过程中的异常处理。当任务抛出未捕获的异常时，默认情况下线程池不会对其进行处理，这可能导致任务中断但线程池仍然继续运行，从而影响整个系统的正确性。可以通过两种方式来处理异常：一是在任务内部使用try - catch块捕获异常并进行处理；二是通过ThreadPoolExecutor的afterExecute方法来处理任务执行后的异常。
    • 例如：
import java.util.concurrent.*;

public class ThreadPoolExceptionHandling {
    public static void main(String[] args) {
        ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
        executorService.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executorService.submit(() -> {
            throw new RuntimeException("Task failed");
        });
        executorService.afterExecute(null, new RuntimeException("Task failed"));
        executorService.shutdown();
    }
}

在上述代码中，通过afterExecute方法可以对任务执行过程中抛出的异常进行统一处理，增强了线程池的健壮性。

  1. 线程池与系统资源的平衡

    • 线程池的使用需要与系统资源相平衡。如果线程池设置过大，会消耗过多的系统资源，如内存、CPU等，导致系统性能下降甚至崩溃。例如，创建过多的线程会导致大量的上下文切换开销，占用过多的内存空间。
    • 相反，如果线程池设置过小，任务会在队列中等待过长时间，导致响应时间变长，无法充分利用系统资源。因此，在创建线程池时，需要根据系统的硬件配置（如CPU核心数、内存大小）和任务的特性（如任务类型、任务执行时间）来合理设置线程池的参数，以达到系统资源的最优利用。

  2. 线程池的扩展性设计

    • 在设计系统时，需要考虑线程池的扩展性。随着业务的发展和任务量的增加，可能需要动态调整线程池的参数。一些高级的线程池实现，如ThreadPoolExecutor，提供了方法来动态调整核心线程数和最大线程数。例如，可以通过setCorePoolSize和setMaxPoolSize方法在运行时根据系统负载情况调整线程池的大小。
    • 另外，还可以考虑使用线程池的集群方案，通过分布式的方式处理大量任务，提高系统的整体处理能力和扩展性。例如，在大型分布式系统中，可以使用多个线程池节点来共同处理任务，通过负载均衡机制将任务分配到不同的节点上，实现系统的高可用性和扩展性。

综上所述，Java Executors提供的不同线程池创建策略各有特点，在实际应用中需要根据任务特性、系统资源等多方面因素进行综合考虑和选择，并通过监控和调优等手段对线程池进行优化，以实现高效、稳定的多线程编程。