咱们先聊聊 ForkJoinPool
你好呀！在 Java 并发编程的世界里，ForkJoinPool 可是个好东西，特别适合处理那些可以“分而治之”的任务。简单来说，它就像一个任务分解大师，能把一个大任务拆成若干个小任务，并行处理，然后再把结果合并起来。这样一来，就能充分利用多核 CPU 的优势，大大提高程序的执行效率。
不过，今天要跟你聊的，可不是 ForkJoinPool 的基本用法，而是更高级的玩法——自定义线程工厂和拒绝策略。相信我，掌握了这些，你就能更灵活地控制 ForkJoinPool 的行为，让它更好地为你的程序服务。
为什么要自定义？
你可能会问，ForkJoinPool 不是挺好用的吗，为啥还要自定义呢？
这就好比买衣服，虽然成衣也能穿，但有时候总觉得差点意思，要么颜色不喜欢，要么款式不合身。这时候，定制就能派上用场了，你可以根据自己的喜好，选择面料、颜色、款式，打造一件独一无二的衣服。
ForkJoinPool 也是一样，虽然它提供了默认的线程工厂和拒绝策略，但在某些特殊场景下，这些默认配置可能无法满足我们的需求。比如：
- 监控与调试： 默认的线程工厂创建的线程，名称都是 ForkJoinPool-worker-1、ForkJoinPool-worker-2 这样的，不方便我们监控和调试。如果能自定义线程名称，加上一些业务标识，那就能一眼看出哪个线程出了问题。
- 资源隔离： 不同的 ForkJoinPool 实例可能会共享同一个线程池，如果其中一个实例提交了大量耗时任务，可能会影响其他实例的性能。如果能为每个实例定制独立的线程池，就能实现资源隔离。
- 自定义拒绝策略：当提交的任务过多时，线程池无法再接收任务，那么就会触发拒绝策略，默认策略一般为抛出异常。在某些场景下，我们可能需要更温和的处理策略，比如将任务保存到队列中稍后执行、记录日志等。
自定义线程工厂
怎么自定义？
要自定义线程工厂，我们需要实现 ForkJoinPool.ForkJoinWorkerThreadFactory 接口。这个接口只有一个方法需要实现：
public interface ForkJoinWorkerThreadFactory {
ForkJoinWorkerThread newThread(ForkJoinPool pool);
}
这个方法接收一个 ForkJoinPool 对象作为参数，返回一个 ForkJoinWorkerThread 对象。ForkJoinWorkerThread 是 ForkJoinPool 中工作线程的基类，我们可以继承它，并在构造函数中做一些自定义操作，比如设置线程名称、优先级、是否为守护线程等。
举个栗子
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinWorkerThread;
import java.util.concurrent.atomic.AtomicInteger;
public class MyForkJoinWorkerThreadFactory implements ForkJoinPool.ForkJoinWorkerThreadFactory {
private static final AtomicInteger threadNumber = new AtomicInteger(1);
private final String threadNamePrefix;
public MyForkJoinWorkerThreadFactory(String threadNamePrefix) {
this.threadNamePrefix = threadNamePrefix;
}
@Override
public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return new MyForkJoinWorkerThread(pool, threadNamePrefix + "-" + threadNumber.getAndIncrement());
}
static class MyForkJoinWorkerThread extends ForkJoinWorkerThread {
protected MyForkJoinWorkerThread(ForkJoinPool pool, String name) {
super(pool);
setName(name);
// 还可以设置其他属性，比如优先级、是否为守护线程等
}
}
}
在这个例子里，我们创建了一个自定义的线程工厂 MyForkJoinWorkerThreadFactory。它接收一个 threadNamePrefix 参数，用于设置线程名称的前缀。在 newThread 方法中，我们创建了一个 MyForkJoinWorkerThread 对象，并为它设置了一个带有前缀和序号的名称。
怎么用？
创建好自定义的线程工厂后，我们可以在创建 ForkJoinPool 实例时，将它作为参数传入：
ForkJoinPool pool = new ForkJoinPool(4, new MyForkJoinWorkerThreadFactory("my-pool"), null, false);
这样，这个 ForkJoinPool 实例就会使用我们自定义的线程工厂来创建线程了。
自定义拒绝策略
怎么自定义？
要自定义拒绝策略，我们需要实现 RejectedExecutionHandler 接口。这个接口只有一个方法需要实现：
public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
这个方法接收两个参数：
r：被拒绝的任务。executor：当前的ThreadPoolExecutor对象（ForkJoinPool 内部使用了ThreadPoolExecutor）。
我们可以在这个方法中实现自己的拒绝策略，比如：
- 抛出异常（默认策略）。
- 将任务保存到队列中稍后执行。
- 记录日志。
- 直接丢弃任务。
- ... 等等
举个栗子
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
public class MyRejectedExecutionHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
// 记录日志
System.err.println("Task rejected: " + r.toString() + ", executor: " + executor.toString());
// 还可以做其他处理，比如将任务保存到队列中稍后执行
}
}
在这个例子里，我们创建了一个自定义的拒绝策略 MyRejectedExecutionHandler。在 rejectedExecution 方法中，我们只是简单地将拒绝的任务和线程池信息记录到日志中。你可以根据自己的需求，实现更复杂的处理逻辑。
怎么用？
创建好自定义的拒绝策略后，我们可以在创建 ForkJoinPool 实例时，将它作为参数传入：
//注意，这里我们直接使用了ThreadPoolExecutor的构造方法
//因为ForkJoinPool的构造方法中没有直接设置RejectedExecutionHandler的地方
//但我们可以通过设置uncaughtExceptionHandler间接实现
Thread.UncaughtExceptionHandler handler = (t, e) -> {
if (e instanceof RejectedExecutionException) {
new MyRejectedExecutionHandler().rejectedExecution(((ForkJoinTask<?>)((ForkJoinPool) t.getThreadGroup()).getQueuedTask()),null );
}
};
ForkJoinPool pool = new ForkJoinPool(4, new MyForkJoinWorkerThreadFactory("my-pool"), handler, false);
总结一下
今天，我们一起学习了如何自定义 ForkJoinPool 的线程工厂和拒绝策略。通过自定义线程工厂，我们可以更好地控制线程的创建过程，比如设置线程名称、优先级、是否为守护线程等。通过自定义拒绝策略，我们可以更灵活地处理被拒绝的任务，比如记录日志、将任务保存到队列中稍后执行等。
希望这些知识能帮助你更好地使用 ForkJoinPool，让它成为你并发编程的得力助手！
还有一些高级技巧
- 动态调整线程池大小： ForkJoinPool 提供了
setParallelism()方法，可以动态调整线程池的大小。这在某些场景下非常有用，比如根据负载情况动态调整线程数量。 - 监控线程池状态： ForkJoinPool 提供了一些方法，可以获取线程池的状态信息，比如
getPoolSize()、getActiveThreadCount()、getRunningThreadCount()、getQueuedTaskCount()等。我们可以利用这些方法来监控线程池的运行情况。 - 使用 CompletableFuture： CompletableFuture 是 Java 8 引入的一个强大的异步编程工具，它可以与 ForkJoinPool 结合使用，实现更复杂的并发编程模式。如果你还不了解 CompletableFuture，强烈建议你去学习一下。
最后的最后
并发编程是一个充满挑战的领域，ForkJoinPool 只是其中的一个工具。要想真正掌握并发编程，还需要不断学习、不断实践。希望这篇文章能为你提供一些帮助，祝你在并发编程的道路上越走越远！
如果你还有其他问题，或者想了解更多关于 ForkJoinPool 的知识，欢迎随时提问。我会尽力解答你的疑惑。
加油！
