面试题：Java线程池如何处理任务拒绝

Question

在Java线程池中，当任务无法被立即执行（例如线程池已满且工作队列也已满）时会发生任务拒绝。请描述Java线程池提供了哪些内置的拒绝策略，它们的实现原理是什么，以及在实际应用场景中如何根据需求选择合适的拒绝策略？同时，如果内置策略都不满足需求，如何自定义拒绝策略？

Answer 1

内置拒绝策略及其实现原理

1. AbortPolicy（默认策略）
   • 原理：当任务无法被执行时，直接抛出RejectedExecutionException异常，阻止系统正常运行。
   • 代码示例：在ThreadPoolExecutor中，当execute方法发现无法处理任务时，如果采用此策略则抛出异常。
2. CallerRunsPolicy
   • 原理：当任务被拒绝添加到线程池时，该任务会在调用execute方法的线程中被执行。这样做的目的是为了减少新任务提交的速度，给线程池中的任务一些执行的时间，避免任务堆积。
   • 代码示例：ThreadPoolExecutor的rejectedExecution方法中，采用此策略时，会直接在调用线程中执行runnable.run()。
3. DiscardPolicy
   • 原理：当任务被拒绝时，直接丢弃该任务，不做任何处理。
   • 代码示例：ThreadPoolExecutor的rejectedExecution方法中，如果采用此策略，代码为空，直接丢弃任务。
4. DiscardOldestPolicy
   • 原理：当任务被拒绝时，会丢弃工作队列中最老的一个任务（即队列头部的任务），然后尝试重新提交被拒绝的任务。
   • 代码示例：ThreadPoolExecutor的rejectedExecution方法中，采用此策略时，会先移除队列头部任务，再尝试提交当前任务。

实际应用场景中选择合适的拒绝策略

1. AbortPolicy
   • 适用场景：适用于对任务执行失败非常敏感的场景，例如一些关键业务流程，一旦任务执行失败需要立即知晓并处理。如金融交易系统，若交易任务无法执行，抛出异常可以让相关人员及时处理。
2. CallerRunsPolicy
   • 适用场景：适用于处理任务提交速率过快，且希望通过调用线程分担任务执行压力的场景。比如在一个简单的日志记录系统中，当线程池处理不过来时，让调用线程帮忙记录日志，不会丢失日志信息。
3. DiscardPolicy
   • 适用场景：适用于对任务可靠性要求不高，允许部分任务丢失的场景。例如在一些统计数据的采集系统中，偶尔丢失一些统计任务对整体统计结果影响不大。
4. DiscardOldestPolicy
   • 适用场景：适用于希望优先处理新任务，且任务执行顺序不是特别关键的场景。比如在一个实时数据更新系统中，新的数据更重要，丢弃旧任务以处理新任务。

自定义拒绝策略

1. 实现方式：实现RejectedExecutionHandler接口，并重写rejectedExecution方法。
2. 代码示例

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 自定义处理逻辑，例如记录日志、尝试将任务放入其他队列等
        System.out.println("任务 " + r + " 被拒绝，自定义处理逻辑...");
    }
}

3. 使用示例

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

public class CustomRejectionPolicyExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, 
                4, 
                10, 
                TimeUnit.SECONDS, 
                new LinkedBlockingQueue<>(2), 
                new CustomRejectedExecutionHandler());
        for (int i = 0; i < 10; i++) {
            executor.execute(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getName() + " 正在执行任务");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
}

在上述代码中，CustomRejectedExecutionHandler实现了自定义拒绝策略，CustomRejectionPolicyExample展示了如何将自定义拒绝策略应用到ThreadPoolExecutor中。