Java 线程池避免共享线程池问题的策略
2021-02-057.5k 阅读
Java 线程池避免共享线程池问题的策略
线程池基础概念回顾
在深入探讨如何避免共享线程池问题之前，我们先来回顾一下 Java 线程池的基础概念。Java 线程池是一种管理和复用线程的机制，其核心类是 ThreadPoolExecutor。通过线程池，我们可以控制线程的创建数量、线程的生命周期以及任务的执行方式，从而提高系统的性能和资源利用率。
ThreadPoolExecutor 有几个关键参数：
- corePoolSize：核心线程数，线程池在初始化后会创建
corePoolSize数量的线程，这些线程即使在空闲时也不会被销毁（除非设置了allowCoreThreadTimeOut为true）。 - maximumPoolSize：最大线程数，线程池所能容纳的最大线程数。当任务队列已满且活跃线程数小于
maximumPoolSize时，线程池会创建新的线程来执行任务。 - keepAliveTime：非核心线程的存活时间，当线程数超过
corePoolSize时，多余的非核心线程如果在keepAliveTime时间内没有任务可执行，就会被销毁。 - unit：
keepAliveTime的时间单位。 - workQueue：任务队列，用于存放提交但尚未被执行的任务。常见的任务队列有
ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue等。 - threadFactory：线程工厂，用于创建新的线程，通过线程工厂可以设置线程的名称、优先级等属性。
- handler：拒绝策略，当任务队列已满且线程数达到
maximumPoolSize时，新提交的任务会被拒绝，由拒绝策略来决定如何处理这些被拒绝的任务。常见的拒绝策略有AbortPolicy（抛出异常）、CallerRunsPolicy（在调用者线程中执行任务）、DiscardPolicy（直接丢弃任务）和DiscardOldestPolicy（丢弃队列中最老的任务，然后尝试提交新任务）。
以下是一个简单创建 ThreadPoolExecutor 的代码示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
public static void main(String[] args) {
// 创建任务队列
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
// 创建线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
5, // corePoolSize
10, // maximumPoolSize
10, // keepAliveTime
TimeUnit.SECONDS,
workQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
// 提交任务
for (int i = 0; i < 20; i++) {
executor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is running task.");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
// 关闭线程池
executor.shutdown();
}
}
在上述代码中，我们创建了一个核心线程数为 5，最大线程数为 10，任务队列容量为 10 的线程池。并向线程池中提交了 20 个任务，由于任务数量超过了核心线程数和任务队列容量之和，部分任务会创建新的线程来执行，而当线程数达到最大线程数 10 后，剩余任务会根据拒绝策略在调用者线程中执行。
共享线程池带来的问题
- 资源竞争 在多个模块或业务逻辑中共享同一个线程池时，不同任务可能会竞争线程池中的资源。例如，一个模块中的任务可能是 CPU 密集型的，而另一个模块中的任务可能是 I/O 密集型的。如果这些任务都在共享的线程池中执行，CPU 密集型任务可能会占用大量的 CPU 资源，导致 I/O 密集型任务得不到及时执行，反之亦然。这会使得整个系统的性能受到影响。
假设我们有两个类型的任务，一个是计算密集型任务，另一个是 I/O 密集型任务，共享同一个线程池：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class SharedThreadPoolResourceCompetition {
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 < 5; i++) {
executor.submit(() -> {
long sum = 0;
for (long j = 0; j < 1000000000L; j++) {
sum += j;
}
System.out.println(Thread.currentThread().getName() + " finished CPU - intensive task.");
});
}
// 提交 I/O 密集型任务
for (int i = 0; i < 5; i++) {
executor.submit(() -> {
try {
Thread.sleep(2000);
System.out.println(Thread.currentThread().getName() + " finished I/O - intensive task.");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executor.shutdown();
}
}
在这个例子中，计算密集型任务会占用大量 CPU 时间，I/O 密集型任务可能需要等待较长时间才能执行，导致整体效率低下。
- 任务优先级混乱 不同模块的任务可能具有不同的优先级。在共享线程池中，由于所有任务都在同一个线程池队列中排队，很难保证高优先级的任务能够优先执行。这可能会导致关键业务逻辑的延迟，影响系统的整体可用性。
例如，我们有一个订单处理系统和一个日志记录系统共享同一个线程池。订单处理任务显然应该具有更高的优先级，但在共享线程池中，日志记录任务可能会先于订单处理任务执行，从而延迟订单的处理。
- 故障蔓延 如果一个模块中的任务出现问题，例如抛出未处理的异常，在共享线程池中，这个问题可能会影响到其他模块的任务执行。由于线程池中的线程是共享的，一个任务的异常可能导致线程终止，进而影响到整个线程池的稳定性，其他模块的任务也可能因此无法正常执行。
假设在共享线程池中有一个任务会抛出异常：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class SharedThreadPoolFailurePropagation {
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());
// 提交一个会抛出异常的任务
executor.submit(() -> {
throw new RuntimeException("Task failed.");
});
// 提交其他正常任务
for (int i = 0; i < 5; i++) {
executor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is running normal task.");
});
}
executor.shutdown();
}
}
在这个例子中，抛出异常的任务可能会导致线程终止，从而影响其他正常任务的执行。
避免共享线程池问题的策略
- 独立线程池 为每个模块或业务逻辑创建独立的线程池是避免共享线程池问题的最直接方法。这样每个模块都有自己独立的线程资源，不会与其他模块产生资源竞争。同时，每个线程池可以根据自身任务的特点来设置合适的参数，如核心线程数、最大线程数、任务队列等。
例如，对于一个电商系统，我们可以为订单处理模块、库存管理模块和用户推荐模块分别创建独立的线程池：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class SeparateThreadPoolExample {
public static void main(String[] args) {
// 订单处理线程池
BlockingQueue<Runnable> orderQueue = new LinkedBlockingQueue<>(10);
ThreadPoolExecutor orderExecutor = new ThreadPoolExecutor(
3,
5,
5,
TimeUnit.SECONDS,
orderQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
// 库存管理线程池
BlockingQueue<Runnable> inventoryQueue = new LinkedBlockingQueue<>(10);
ThreadPoolExecutor inventoryExecutor = new ThreadPoolExecutor(
2,
4,
5,
TimeUnit.SECONDS,
inventoryQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
// 用户推荐线程池
BlockingQueue<Runnable> recommendationQueue = new LinkedBlockingQueue<>(10);
ThreadPoolExecutor recommendationExecutor = new ThreadPoolExecutor(
4,
6,
5,
TimeUnit.SECONDS,
recommendationQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
// 提交订单处理任务
orderExecutor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is processing order.");
});
// 提交库存管理任务
inventoryExecutor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is managing inventory.");
});
// 提交用户推荐任务
recommendationExecutor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is generating recommendation.");
});
// 关闭线程池
orderExecutor.shutdown();
inventoryExecutor.shutdown();
recommendationExecutor.shutdown();
}
}
通过这种方式，不同模块的任务在各自独立的线程池中执行，避免了资源竞争、任务优先级混乱和故障蔓延的问题。
- 线程池隔离框架 除了手动创建独立线程池，还可以使用一些线程池隔离框架来实现类似的功能。例如，Hystrix 是 Netflix 开源的一个容错框架，它提供了线程池隔离的功能。通过 Hystrix，我们可以将不同的服务或业务逻辑隔离在不同的线程池中，从而避免相互影响。
使用 Hystrix 实现线程池隔离的示例代码如下： 首先，需要引入 Hystrix 的依赖：
<dependency>
<groupId>com.netflix.hystrix</groupId>
<artifactId>hystrix-core</artifactId>
<version>1.5.18</version>
</dependency>
然后，定义一个 Hystrix 命令类：
import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
public class HystrixThreadPoolIsolationCommand extends HystrixCommand<String> {
public HystrixThreadPoolIsolationCommand() {
super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
}
@Override
protected String run() throws Exception {
// 实际业务逻辑
return "Task executed successfully.";
}
}
在主程序中使用该命令：
public class HystrixExample {
public static void main(String[] args) {
HystrixThreadPoolIsolationCommand command = new HystrixThreadPoolIsolationCommand();
String result = command.execute();
System.out.println(result);
}
}
Hystrix 会为每个命令组创建独立的线程池，从而实现线程池隔离，避免共享线程池带来的问题。
- 使用优先级队列
如果无法完全避免共享线程池，可以通过使用优先级队列来部分解决任务优先级混乱的问题。Java 中的
PriorityBlockingQueue是一个支持优先级的阻塞队列。我们可以自定义任务类实现Comparable接口，根据任务的优先级进行排序。
以下是一个使用 PriorityBlockingQueue 的示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
class PriorityTask implements Comparable<PriorityTask> {
private int priority;
private String taskName;
public PriorityTask(int priority, String taskName) {
this.priority = priority;
this.taskName = taskName;
}
@Override
public int compareTo(PriorityTask other) {
return this.priority - other.priority;
}
@Override
public String toString() {
return "PriorityTask{" +
"priority=" + priority +
", taskName='" + taskName + '\'' +
'}';
}
public void execute() {
System.out.println(Thread.currentThread().getName() + " is executing " + this);
}
}
public class PriorityThreadPoolExample {
public static void main(String[] args) {
BlockingQueue<PriorityTask> workQueue = new PriorityBlockingQueue<>();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
3,
5,
5,
TimeUnit.SECONDS,
workQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
// 提交高优先级任务
executor.submit(() -> new PriorityTask(1, "High - priority task").execute());
// 提交低优先级任务
executor.submit(() -> new PriorityTask(3, "Low - priority task").execute());
// 提交中优先级任务
executor.submit(() -> new PriorityTask(2, "Medium - priority task").execute());
executor.shutdown();
}
}
在这个示例中，任务会按照优先级顺序在共享线程池中执行，一定程度上解决了任务优先级混乱的问题。
- 异常处理与监控 对于共享线程池中的任务异常，要进行妥善处理。可以在任务执行过程中捕获异常，并进行适当的日志记录和处理，避免异常导致线程终止影响其他任务。同时，要对线程池进行监控，及时发现线程池中的异常情况，如线程数的变化、任务队列的长度等。
Java 提供了一些工具类来实现线程池的监控，例如 ThreadPoolExecutor 类本身就提供了一些获取线程池状态的方法，如 getActiveCount（获取当前活动线程数）、getQueue（获取任务队列）等。
以下是一个简单的异常处理和监控示例：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExceptionHandlingAndMonitoring {
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());
// 提交任务并处理异常
executor.submit(() -> {
try {
// 模拟任务执行
Thread.sleep(1000);
if (Math.random() < 0.5) {
throw new RuntimeException("Task failed randomly.");
}
System.out.println(Thread.currentThread().getName() + " finished task successfully.");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (RuntimeException e) {
System.err.println(Thread.currentThread().getName() + " encountered an error: " + e.getMessage());
}
});
// 监控线程池
Thread monitorThread = new Thread(() -> {
while (true) {
System.out.println("Active threads: " + executor.getActiveCount());
System.out.println("Queue size: " + executor.getQueue().size());
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
});
monitorThread.setDaemon(true);
monitorThread.start();
executor.shutdown();
}
}
在这个示例中，我们在任务中捕获了异常并进行了日志记录。同时，通过一个守护线程定期监控线程池的活动线程数和任务队列长度，以便及时发现问题。
总结
共享线程池在多模块或业务逻辑中使用时可能会带来资源竞争、任务优先级混乱和故障蔓延等问题。为了避免这些问题，可以采用独立线程池、线程池隔离框架、使用优先级队列以及加强异常处理与监控等策略。在实际应用中，需要根据具体的业务场景和需求，选择合适的策略来优化线程池的使用，提高系统的性能和稳定性。
通过合理地设计和管理线程池，我们能够充分发挥多线程编程的优势，让 Java 应用程序在处理高并发任务时更加高效和可靠。希望以上内容能帮助你在 Java 开发中更好地避免共享线程池带来的问题，构建健壮的多线程应用程序。