Java 线程池与周期性任务
2021-07-096.8k 阅读
Java 线程池基础
在 Java 多线程编程中，线程池是一种非常重要的工具。线程池管理着一组工作线程，它的主要目的是复用已有的线程，而不是每次需要执行任务时都创建新的线程。这样做有几个显著的优点：
- 降低资源消耗：创建和销毁线程是比较消耗系统资源的操作。线程池通过复用线程，减少了这种开销。例如，在一个高并发的 Web 服务器应用中，如果每个请求都创建新线程处理，当请求量很大时，频繁的线程创建和销毁会严重影响系统性能。而使用线程池，这些线程可以被重复使用，大大降低了资源消耗。
- 提高响应速度：由于线程已经预先创建好，当有新任务到达时，无需等待线程创建，直接可以分配线程执行任务，从而提高了系统的响应速度。比如在一个实时数据处理系统中，快速的响应对于数据的及时处理和分析至关重要，线程池能够很好地满足这一需求。
- 方便线程管理：线程池提供了一种统一管理线程的方式，可以控制线程的数量，设置线程的优先级等。例如，可以根据系统的负载情况动态调整线程池中的线程数量，避免过多线程导致系统资源耗尽，或者过少线程导致任务处理不及时。
线程池的创建
在 Java 中，可以通过 ExecutorService 接口及其实现类来创建线程池。Executors 类提供了一些静态方法来方便地创建不同类型的线程池。
- FixedThreadPool：创建一个固定大小的线程池，线程池中的线程数量是固定的。当有新任务提交时，如果线程池中有空闲线程，则直接使用空闲线程执行任务；如果没有空闲线程，则任务会被放入队列中等待，直到有线程完成任务并变为空闲。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FixedThreadPoolExample {
public static void main(String[] args) {
// 创建一个固定大小为 3 的线程池
ExecutorService executorService = Executors.newFixedThreadPool(3);
for (int i = 0; i < 5; i++) {
final 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 个任务会在队列中等待，直到有线程完成任务后空闲出来。 2. CachedThreadPool：创建一个可缓存的线程池，如果线程池中的线程空闲时间超过 60 秒，这些线程会被回收。当有新任务提交时，如果线程池中有空闲线程，则使用空闲线程执行任务；如果没有空闲线程，则创建一个新线程来执行任务。
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++) {
final 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 个任务时，由于一开始没有空闲线程，会创建 5 个新线程来执行任务。随着任务执行完毕，线程如果在 60 秒内没有新任务分配，就会被回收。 3. SingleThreadExecutor：创建一个单线程的线程池，只有一个线程在工作。所有任务按照提交的顺序依次执行，保证了任务执行的顺序性。
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++) {
final 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 个任务会依次由唯一的线程执行，确保了任务执行的顺序性。
- 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 scheduledExecutorService = Executors.newScheduledThreadPool(3);
scheduledExecutorService.schedule(() -> {
System.out.println("Delayed task is running on thread " + Thread.currentThread().getName());
}, 5, TimeUnit.SECONDS);
scheduledExecutorService.shutdown();
}
}
上述代码创建了一个大小为 3 的 ScheduledThreadPool，并提交了一个延迟 5 秒执行的任务。
线程池的核心参数
在 ThreadPoolExecutor 类（Executors 创建的线程池底层大多是 ThreadPoolExecutor）中，有几个核心参数决定了线程池的行为。
- corePoolSize：核心线程数，线程池在初始化后会创建
corePoolSize个线程，这些线程会一直存活，即使它们处于空闲状态，除非设置了allowCoreThreadTimeOut为true。例如，在一个数据库连接池应用中，核心线程数可以设置为数据库连接的最小数量，确保始终有一定数量的连接可用。 - maximumPoolSize：最大线程数，线程池允许创建的最大线程数量。当任务队列已满且核心线程都在忙碌时，线程池会创建新的线程，直到线程数量达到
maximumPoolSize。如果此时任务队列仍然满且所有线程都在忙碌，新提交的任务会根据RejectedExecutionHandler策略进行处理。比如在一个高并发的计算任务场景中，根据服务器的资源情况合理设置最大线程数，可以避免系统资源被过度消耗。 - keepAliveTime：非核心线程的存活时间。当线程池中的线程数量超过
corePoolSize时，多余的非核心线程如果在keepAliveTime时间内没有任务执行，就会被回收。例如，在一个 Web 应用中，当请求量高峰期过后，多余的线程在一段时间没有新请求处理后就会被回收，以节省系统资源。 - unit：
keepAliveTime的时间单位，如TimeUnit.SECONDS、TimeUnit.MILLISECONDS等。 - workQueue：任务队列，用于存放等待执行的任务。常见的任务队列有
ArrayBlockingQueue（有界队列）、LinkedBlockingQueue（无界队列）、SynchronousQueue（不存储任务，直接提交给线程执行）等。例如，ArrayBlockingQueue适用于需要严格控制任务数量的场景，而LinkedBlockingQueue适用于任务量较大且不希望拒绝任务的场景。 - threadFactory：线程工厂，用于创建线程池中的线程。可以通过自定义线程工厂来设置线程的名称、优先级等属性。例如，在一个多模块的应用中，可以通过自定义线程工厂为不同模块的线程设置不同的名称前缀，方便调试和监控。
- handler：拒绝策略，当任务队列已满且线程池中的线程数量达到
maximumPoolSize时，新提交的任务会被拒绝，此时会根据RejectedExecutionHandler策略进行处理。常见的拒绝策略有AbortPolicy（抛出RejectedExecutionException异常）、CallerRunsPolicy（由提交任务的线程来执行任务）、DiscardPolicy（直接丢弃任务）、DiscardOldestPolicy（丢弃队列中最老的任务，然后尝试提交新任务）。
深入理解 Java 线程池的工作原理
线程池的工作流程大致如下：
- 当一个任务提交到线程池时，线程池首先检查当前运行的线程数是否小于
corePoolSize。如果是，则创建一个新线程来执行任务，即使此时有其他空闲线程。 - 如果当前运行的线程数已经达到
corePoolSize，则将任务放入任务队列workQueue中等待。 - 如果任务队列已满，且当前运行的线程数小于
maximumPoolSize，则创建新的线程来执行任务。 - 如果任务队列已满且当前运行的线程数已经达到
maximumPoolSize，则根据拒绝策略handler来处理新提交的任务。
例如，假设我们有一个线程池，corePoolSize 为 2，maximumPoolSize 为 4，任务队列是一个大小为 3 的 ArrayBlockingQueue。当有 10 个任务依次提交时：
- 前 2 个任务会立即由新创建的 2 个核心线程执行。
- 接下来 3 个任务会放入任务队列中等待。
- 再接下来 2 个任务由于任务队列已满，且当前线程数小于
maximumPoolSize，会创建 2 个新的非核心线程来执行。 - 此时线程池中有 4 个线程在运行，任务队列已满。当再提交任务时，根据设置的拒绝策略进行处理，如果是
AbortPolicy，则会抛出RejectedExecutionException异常。
Java 中的周期性任务
在很多实际应用场景中，我们需要执行周期性任务，比如定时备份数据、定时清理缓存等。Java 提供了几种方式来实现周期性任务，其中使用 ScheduledThreadPool 是一种非常灵活且高效的方式。
ScheduledThreadPool 的使用
ScheduledThreadPool 继承自 ThreadPoolExecutor 并实现了 ScheduledExecutorService 接口。它提供了两个主要方法来执行周期性任务：
- scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)：该方法会在指定的初始延迟
initialDelay后开始执行任务，然后以固定的周期period重复执行任务。无论任务执行时间长短，都会按照固定的周期启动下一次执行。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheduleAtFixedRateExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Task is running on thread " + Thread.currentThread().getName() + " at " + System.currentTimeMillis());
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 1, 2, TimeUnit.SECONDS);
}
}
在上述代码中，任务会在延迟 1 秒后开始执行，然后每隔 2 秒执行一次。由于任务执行时间为 3 秒，大于周期 2 秒，所以下一次执行会在前一次执行完成后立即启动，而不会等待 2 秒的间隔。
2. scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)：该方法同样在指定的初始延迟 initialDelay 后开始执行任务，但后续的执行是在上一次任务执行完成后，间隔 delay 时间再启动下一次执行。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheduleWithFixedDelayExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
scheduledExecutorService.scheduleWithFixedDelay(() -> {
System.out.println("Task is running on thread " + Thread.currentThread().getName() + " at " + System.currentTimeMillis());
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 1, 2, TimeUnit.SECONDS);
}
}
此代码中，任务延迟 1 秒后开始执行，每次执行完成后，间隔 2 秒再启动下一次执行。由于任务执行时间为 3 秒，所以实际间隔是 3 + 2 = 5 秒执行一次。
周期性任务的应用场景
- 数据备份：在数据库应用中，可以使用周期性任务定时备份数据库。例如，每天凌晨 2 点进行一次全量备份，每小时进行一次增量备份。通过
ScheduledThreadPool可以方便地设置这些备份任务的执行时间和周期。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class DatabaseBackupExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
// 每天凌晨 2 点进行全量备份
long initialDelayFullBackup = calculateInitialDelay(2, 0, 0);
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Full database backup is running on thread " + Thread.currentThread().getName());
// 实际备份逻辑在此处添加
}, initialDelayFullBackup, 24, TimeUnit.HOURS);
// 每小时进行增量备份
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Incremental database backup is running on thread " + Thread.currentThread().getName());
// 实际备份逻辑在此处添加
}, 0, 1, TimeUnit.HOURS);
}
private static long calculateInitialDelay(int hour, int minute, int second) {
long currentTime = System.currentTimeMillis();
long targetTime = currentTime / 1000;
targetTime -= targetTime % (24 * 60 * 60);
targetTime += hour * 60 * 60 + minute * 60 + second;
if (currentTime / 1000 >= targetTime) {
targetTime += 24 * 60 * 60;
}
return (targetTime - currentTime / 1000) * 1000;
}
}
- 缓存清理：在 Web 应用中，缓存的数据可能会随着时间推移变得不再有效，需要定时清理。可以使用周期性任务每隔一段时间检查并清理过期的缓存数据。
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class CacheCleaningExample {
private static Map<String, Long> cache = new HashMap<>();
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
// 模拟缓存数据
cache.put("key1", System.currentTimeMillis());
cache.put("key2", System.currentTimeMillis());
scheduledExecutorService.scheduleAtFixedRate(() -> {
long currentTime = System.currentTimeMillis();
cache.entrySet().removeIf(entry -> currentTime - entry.getValue() > 60 * 1000);
System.out.println("Cache cleaning is running on thread " + Thread.currentThread().getName() + ", current cache size: " + cache.size());
}, 0, 5, TimeUnit.MINUTES);
}
}
- 监控系统：在分布式系统中，需要定时收集各个节点的状态信息，如 CPU 使用率、内存使用率等。通过周期性任务可以按照设定的时间间隔去各个节点采集数据并进行分析。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class MonitoringSystemExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
// 假设系统中有 5 个节点
for (int i = 0; i < 5; i++) {
final int nodeIndex = i;
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Collecting data from node " + nodeIndex + " on thread " + Thread.currentThread().getName());
// 实际数据采集逻辑在此处添加
}, 0, 1, TimeUnit.MINUTES);
}
}
}
处理周期性任务中的异常
在执行周期性任务时，异常处理是非常重要的。如果在任务执行过程中抛出异常，默认情况下，ScheduledThreadPool 不会捕获异常，这可能导致任务停止执行。为了避免这种情况，可以在任务中添加异常处理逻辑。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ExceptionHandlingInScheduledTask {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
scheduledExecutorService.scheduleAtFixedRate(() -> {
try {
System.out.println("Task is running on thread " + Thread.currentThread().getName());
throw new RuntimeException("Simulated exception");
} catch (Exception e) {
System.out.println("Exception caught: " + e.getMessage());
}
}, 0, 2, TimeUnit.SECONDS);
}
}
在上述代码中，任务会抛出一个模拟的运行时异常，但由于在任务内部捕获了异常，任务不会停止执行，仍然会按照周期继续运行。
另外，也可以通过自定义 Thread.UncaughtExceptionHandler 来处理未捕获的异常。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class CustomUncaughtExceptionHandlerExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
Thread.setDefaultUncaughtExceptionHandler((thread, throwable) -> {
System.out.println("Uncaught exception in thread " + thread.getName() + ": " + throwable.getMessage());
});
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Task is running on thread " + Thread.currentThread().getName());
throw new RuntimeException("Simulated uncaught exception");
}, 0, 2, TimeUnit.SECONDS);
}
}
在这个例子中，通过设置默认的 UncaughtExceptionHandler，当任务抛出未捕获的异常时，会打印异常信息，同时任务不会停止执行。
优化和调优线程池与周期性任务
- 合理设置线程池参数：根据应用的实际需求和系统资源情况，合理设置线程池的
corePoolSize、maximumPoolSize等参数。例如，对于 CPU 密集型任务，线程池大小可以设置为 CPU 核心数加 1，以充分利用 CPU 资源并避免线程上下文切换开销；对于 I/O 密集型任务，由于线程大部分时间处于等待 I/O 操作完成的状态，可以适当增加线程池大小，以提高系统的并发处理能力。 - 选择合适的任务队列：根据任务的特点选择合适的任务队列。如果任务量相对稳定且不希望任务被拒绝，可以选择
LinkedBlockingQueue；如果需要严格控制任务数量，防止任务堆积导致内存溢出，可以选择ArrayBlockingQueue；如果希望任务能够尽快执行，不进行排队，可以选择SynchronousQueue。 - 监控和分析：使用工具如 JConsole、VisualVM 等对线程池和周期性任务进行监控，分析线程池的运行状态，如线程利用率、任务队列长度等。根据监控数据进行调优，确保系统性能最优。
- 资源管理：在执行周期性任务时，要注意资源的合理使用和释放。例如，在进行数据库备份任务时，确保数据库连接在任务完成后及时关闭，避免资源泄漏。
线程池与周期性任务的高级应用
- 分布式任务调度：在分布式系统中，可以使用线程池和周期性任务来实现分布式任务调度。例如，使用
ScheduledThreadPool在各个节点上定时执行任务，并通过分布式协调工具（如 ZooKeeper）来保证任务在整个集群中的一致性和唯一性。 - 动态任务调整：根据系统的运行状态动态调整线程池的参数和周期性任务的执行周期。例如，当系统负载较高时，适当减少周期性任务的执行频率，或者增加线程池的大小以提高任务处理能力；当系统负载较低时，反之操作。
- 任务优先级：在某些应用场景中，任务可能有不同的优先级。可以通过自定义任务队列和线程池的调度策略，实现任务按照优先级执行。例如，将高优先级任务优先分配给线程执行，确保重要任务能够及时处理。
总结线程池与周期性任务的注意事项
- 线程安全：在多线程环境下，要确保共享资源的访问是线程安全的。对于周期性任务中涉及到的共享数据，要使用合适的同步机制（如
synchronized、Lock等）来保证数据的一致性和完整性。 - 资源限制：要注意系统资源的限制，避免线程池过度消耗资源导致系统性能下降甚至崩溃。合理设置线程池大小和任务队列容量，根据系统的硬件资源（如 CPU、内存等）进行调整。
- 任务依赖：如果周期性任务之间存在依赖关系，要确保任务的执行顺序正确。可以使用
CountDownLatch、CyclicBarrier等工具来协调任务之间的同步。 - 性能测试：在实际应用中，对线程池和周期性任务进行性能测试是非常必要的。通过性能测试，可以发现潜在的性能瓶颈，优化线程池参数和任务执行逻辑，提高系统的整体性能。
通过深入理解和合理使用 Java 线程池与周期性任务，可以有效地提高应用程序的性能、稳定性和可维护性，满足各种复杂的业务需求。在实际开发中，需要根据具体的应用场景，灵活运用线程池和周期性任务的相关知识，进行优化和调优，以达到最佳的效果。