Spring TaskExecutor顺序执行配置、错误处理及自定义队列实现咨询

首先针对你的核心疑问逐一拆解：

1. 当前ThreadPoolTaskExecutor配置的正确性 #

你将corePoolSize和maxPoolSize都设为1的思路是正确的——这样线程池只会维护一个工作线程，所有提交的任务都会进入队列，由这个线程依次取出执行，完全能满足顺序处理通知的需求，从根源避免死锁（单线程不存在资源竞争导致的死锁场景）。

不过你对WaitForTasksToCompleteOnShutdown的理解和配置存在偏差：

• 这个属性的作用是：当Spring上下文销毁时，是否等待线程池中的所有已提交任务执行完成再关闭线程池。
• 它的默认值是false，也就是上下文销毁时会立即中断正在执行的任务，并且丢弃队列中未执行的任务；如果你的需求是让已提交的任务都执行完再关闭，应该把它设为true，同时建议搭配awaitTerminationSeconds属性设置超时时间（比如60），防止线程池无限等待。

修正后的配置示例：

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="1"/>
    <property name="maxPoolSize" value="1"/>
    <property name="waitForTasksToCompleteOnShutdown" value="true"/>
    <property name="awaitTerminationSeconds" value="60"/>
    <!-- 建议添加线程名前缀，方便日志排查 -->
    <property name="threadNamePrefix" value="NotificationProcessor-"/>
    <!-- 建议设置队列容量，避免默认无界队列导致内存溢出 -->
    <property name="queueCapacity" value="1000"/>
</bean>

2. 当前实现的遗漏点 #

• 队列容量未限制：ThreadPoolTaskExecutor默认使用LinkedBlockingQueue，默认容量是Integer.MAX_VALUE，属于无界队列。如果外部通知量突增，队列会无限膨胀，最终导致OOM。一定要显式设置queueCapacity，根据系统承载能力设定合理值。
• 线程命名不规范：未设置threadNamePrefix时，线程名会是类似pool-1-thread-1的默认值，排查问题时很难定位到具体任务线程。
• 异常处理缺失：这是你提到的关键问题，下面单独展开说明。

3. 错误处理的添加位置 #

当前实现中，如果someMethod.processNotification(notification)抛出异常，这个异常会被线程池的工作线程“吃掉”（不会向上抛出），你无法感知任务执行失败。建议从两个层面添加异常处理：

层面一：任务内部的局部处理 #

在提交Runnable时，直接在run()方法中捕获异常，针对当前任务做特定处理：

asynchronousService.executeAsynchronously(new Runnable() {
    @Override
    public void run() {
        try {
            someMethod.processNotification(notification);
        } catch (Exception e) {
            // 记录日志、重试通知、触发告警等
            log.error("处理通知失败，通知ID: {}", notification.getId(), e);
            notificationRetryService.retry(notification);
        }
    }
});

层面二：全局统一异常处理 #

如果希望所有异步任务的异常都有统一处理逻辑，可以在AsynchronousServiceImpl中包装传入的Runnable，添加全局异常捕获：

@Override
public void executeAsynchronously(Runnable task) {
    Runnable wrappedTask = () -> {
        try {
            task.run();
        } catch (Exception e) {
            // 全局异常处理逻辑，比如统一日志记录、告警
            log.error("异步任务执行失败", e);
            alertService.sendAlert("异步任务执行异常", e.getMessage());
        }
    };
    taskExecutor.execute(wrappedTask);
}

这种方式无需每个任务都写重复的try-catch，代码更简洁。

4. 自定义队列与任务优先级的实现 #

完全可以基于Spring的ThreadPoolTaskExecutor实现自定义队列，不需要从零实现ExecutorService，步骤非常简单：

步骤1：定义带优先级的任务类 #

使用PriorityBlockingQueue实现优先级，需要让任务实现Comparable接口，或者给队列传入自定义Comparator：

public class PriorityNotificationTask implements Runnable, Comparable<PriorityNotificationTask> {
    private Notification notification;
    private int priority; // 数字越大优先级越高

    public PriorityNotificationTask(Notification notification, int priority) {
        this.notification = notification;
        this.priority = priority;
    }

    @Override
    public void run() {
        someMethod.processNotification(notification);
    }

    @Override
    public int compareTo(PriorityNotificationTask other) {
        // 降序排列，优先级高的先执行
        return Integer.compare(other.priority, this.priority);
    }
}

步骤2：配置ThreadPoolTaskExecutor使用自定义队列 #

在XML配置中，通过queue属性注入自定义队列：

<bean id="priorityQueue" class="java.util.concurrent.PriorityBlockingQueue">
    <!-- 指定初始容量 -->
    <constructor-arg value="1000"/>
</bean>

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="1"/>
    <property name="maxPoolSize" value="1"/>
    <property name="waitForTasksToCompleteOnShutdown" value="true"/>
    <property name="awaitTerminationSeconds" value="60"/>
    <property name="threadNamePrefix" value="PriorityNotificationProcessor-"/>
    <!-- 注入自定义优先级队列 -->
    <property name="queue" ref="priorityQueue"/>
</bean>

步骤3：提交带优先级的任务 #

asynchronousService.executeAsynchronously(new PriorityNotificationTask(notification, 5));
asynchronousService.executeAsynchronously(new PriorityNotificationTask(urgentNotification, 10)); // 优先级更高，会先被执行

实现任务优先级的难度很低，核心就是利用JDK的PriorityBlockingQueue配合任务排序逻辑，完全不需要脱离Spring的TaskExecutor体系。

内容的提问来源于stack exchange，提问作者Norbert94