“苍生我要救钟离权我也要救”——这句话初看像一句仙侠剧情台词但放在后端系统里它其实就是我们经常遇到的两难局面一边是海量普通任务在不断积压一边是某个核心高优任务必须立即处理。很多同学在处理这种需求时容易走极端要么只保高优任务导致普通任务饿死要么把所有任务一视同仁结果核心任务被长尾任务拖住。本文就围绕这个“两边都要保”的场景拆解一套优先级任务调度方案包含核心原理、完整可运行代码、常见排队问题和工程落地建议。无论你是刚接触并发编程的后端新手还是正在设计任务系统的中台工程师都可以把这段内容当作一份可直接复用的实操笔记。1. 背景与核心概念1.1 从“救苍生”到“任务调度”的映射先把这个标题翻译成技术语言。所谓“救苍生”对应的是业务系统中大量常规任务日志上报、短信通知、数据同步、批量导出、定时对账等。这类任务数量大、单条耗时短、对实时性要求不极端但它们占据了系统绝大部分资源。所谓“救钟离权”则对应一个或少数几个高优核心任务比如用户支付后的订单处理、风控拦截通知、紧急配置刷新、资金动账请求。这类任务数量少但延迟高了就要出事故。问题在于如果只“救苍生”按照先来先服务的方式排队那么一个高优订单处理任务可能排在上千个日志任务后面等轮到它时业务已经超时如果只“救钟离权”来了高优任务就无限插队普通任务可能在极端流量下一直得不到执行最终堆积成雪崩。所以这篇文章要解决的就是如何在同一套任务处理系统里既保证高优任务的低延迟又保证普通任务最终都能被执行不出现饿死。1.2 任务调度系统的基本组成任何任务调度系统不管简单还是复杂都离不开四个核心部分任务输入外部请求把任务提交进系统包括任务ID、优先级、类型、业务数据等。任务队列暂存待执行任务并根据某种策略决定下一个执行谁。任务执行单元真正运行任务的线程或进程。任务生命周期管理从提交、等待、执行、成功、失败、重试到最终结束的完整状态流转。在此基础上优先级任务调度就是在“任务队列”和“执行单元”之间增加一套取舍规则高优任务优先被拿走低优任务不因高优任务持续到达而永久等待。1.3 单机方案与分布式方案的边界在正式写代码之前必须先明确一个边界本文实现的调度器是一个单机进程内的方案适合单体应用、服务内异步任务、后台线程池任务处理等场景。它不直接解决分布式多节点下的全局优先级调度问题。如果你的系统已经引入了消息队列比如 RocketMQ、RabbitMQ、Kafka并且需要在多个消费节点之间做优先级调度那么单机的 PriorityBlockingQueue 就不够用了。截至当前主流实践分布式优先级调度一般会采用“分区主题 按优先级拆分Topic”或“同一Topic内结合消息过滤/延迟消息”等方式实现这部分在最后的最佳实践里会给出建议。不区分单机和分布式就直接讲队列是很多教程容易误导人的地方。我们先稳住单机场景把原理吃透再去拓展分布式。2. 环境准备与版本说明2.1 运行环境本文示例使用 Java 语言和 Spring Boot 框架。之所以选这个组合是因为 Spring Boot 的自动装配和参数配置能让我们把精力集中在调度逻辑上而不是浪费在容器初始化上。你需要准备的基础环境如下JDK 8 及以上版本示例代码用到了 Lambda、Stream 等特性JDK 8 即可。Maven 3.6 以上用于管理依赖。Spring Boot 2.x 或 3.x 均可本文示例以 Spring Boot 2.x 常见写法为主如果你用的是 Spring Boot 3.x依赖坐标中的javax需要替换为jakarta但核心调度代码基本不用改。IDE 推荐 IntelliJ IDEA命令行演示不强制依赖 IDE。版本需要根据你的项目实际情况调整不必照搬某个具体小版本。优先确认你本机已安装 JDK 且java -version能正常输出。2.2 示例项目结构本文实战案例的代码结构如下priority-task-scheduler/ ├── pom.xml └── src/main/ ├── java/com/csdn/task/ │ ├── TaskApplication.java │ ├── model/ │ │ ├── TaskPriority.java │ │ ├── TaskStatus.java │ │ └── Task.java │ ├── queue/ │ │ └── PriorityTaskQueue.java │ ├── producer/ │ │ └── TaskProducer.java │ ├── consumer/ │ │ └── TaskConsumer.java │ ├── retry/ │ │ └── RetryPolicy.java │ └── registry/ │ └── IdempotentTaskRegistry.java └── resources/ └── application.yml这个结构很清晰模型层放任务实体、状态枚举和优先级枚举队列层封装优先级队列生产者负责接收外部任务消费者负责执行重试策略和幂等注册表分别解决失败重试与重复执行问题。2.3 依赖配置在pom.xml中添加 Spring Boot Web 依赖就够了这里用 Web 只是为了提供 HTTP 入口来测试任务提交生产环境中完全可以用业务接口替代。?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.csdn/groupId artifactIdpriority-task-scheduler/artifactId version1.0.0/version packagingjar/packaging parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version relativePath/ /parent properties java.version1.8/java.version /properties dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project如果你的项目已经存在不需要刻意改成这个结构理解每个类的职责后直接迁移到你的工程里即可。3. 核心原理拆解3.1 优先级队列 PriorityBlockingQueueJava 并发包中提供了一个非常合适的数据结构PriorityBlockingQueue。它是一个基于优先级堆的无界阻塞队列线程安全且出队时会按照元素的自然顺序或自定义比较器返回最高优先级的元素。也就是说多个消费者线程同时从队列中take()任务时永远不会拿到“次优”的任务而是拿到当前队列里优先级最高的那个。这一特性天然适合“救钟离权”。但这里有一个关键设计点任务的“优先级”如何定义通常有两种做法用整数表示数字越小优先级越高。用枚举表示通过compareTo方法实现排序。第二种更可读也更容易在业务代码里使用。下面我会采用枚举方式。3.2 高优插队与 low 优先级的饿死问题优先级队列带来的第一个副作用就是“低优先级任务饿死”。如果高优任务源源不断提交进来低优任务可能永远轮不到执行。这在业务上表现为普通日志任务越积越多高优任务持续插队最终普通任务延迟以小时级增长。解决饿死问题业界常用的思路有以下几种任务老化任务在队列中等待超过一定时间后动态提升优先级。权重轮转按优先级权重分配执行名额比如每执行 10 个高优任务至少执行 1 个低优任务。独立队列隔离完全把高低优任务拆到不同线程池避免互相抢占。本文示例采用“任务老化”因为它实现成本低且能直观演示优先级变化过程。3.3 线程池的选择有优先级队列还不够还需要一个线程池来消费任务。Java 的ThreadPoolExecutor支持传入一个BlockingQueue作为工作队列所以我们只需要把PriorityTaskQueue传给ThreadPoolExecutor即可。线程池的核心参数需要认真说明corePoolSize核心线程数常驻线程数量。maximumPoolSize最大线程数任务高峰期最多创建的线程数量。keepAliveTime非核心线程空闲存活时间。unit存活时间的单位。workQueue工作队列必须使用我们的优先级队列。threadFactory线程工厂用来给线程取有意义的名字方便排查日志。handler拒绝策略当线程池和队列都满时触发。拒绝策略是一个非常值得注意的点。默认的AbortPolicy会在队列满时直接抛异常导致任务丢失。对于任务调度系统我更推荐CallerRunsPolicy它会让提交任务的线程自己去执行任务虽然会阻塞生产者但至少任务不会丢。不过CallerRunsPolicy在优先级队列场景下有一个副作用被拒绝回调用线程执行的任务可能会绕过优先级顺序。所以如果你的场景对优先级极其敏感建议不用CallerRunsPolicy而是配合重试机制将失败任务重新投递。3.4 任务状态机一个任务从提交到结束至少要经历以下状态PENDING → RUNNING → SUCCESS ↓ FAILED → RETRYING → RUNNINGPENDING已入队等待执行。RUNNING已从队列取出正在执行。SUCCESS执行成功流程结束。FAILED执行失败进入失败处理流程。RETRYING满足重试条件准备重新入队。为什么要设计这么一套状态机因为只有状态明确才能回答三个关键问题任务是否执行过、执行到一半失败怎么办、重复投递是否安全。很多线上事故都是因为状态不清晰重复消费或失败后静默丢弃。3.5 幂等保障为什么必须考虑任务调度系统在处理“任务丢失”时第一反应往往是“重试”。但重试可能造成重复执行。比如一个订单通知任务第一次执行时网络超时实际上服务端已经收到并处理了重试又发一次用户就收到两条一模一样的通知。解决的办法有两个一是让业务处理逻辑本身支持幂等同一任务执行多次结果一致二是在调度框架层面做去重。框架层面的去重一般依赖任务 ID。本文示例会用ConcurrentHashMap维护一个已执行成功的任务 ID 集合相同 ID 的任务不允许重复执行。这只是单体场景演示分布式场景请使用 Redis 等外部存储做分布式幂等最佳实践部分会提到。4. 完整实战案例优先保障 批量兜底的任务调度器现在进入核心实战。我们的目标是实现这样一个效果普通任务和服务可以随便提交不会被拒绝。高优任务提交后能在 1 秒内被线程取出执行。普通任务即使前面堆了很多高优任务也不会饿死而是等待超过一定时间后自动升级优先级。失败任务最多重试 3 次每次退避时间递增。同一个任务 ID 不会被重复执行。4.1 创建项目结构和启动类先用 Maven 创建工程或者直接在 IDEA 里 Spring Initializr 创建。完成后先写启动类。// 文件路径src/main/java/com/csdn/task/TaskApplication.java package com.csdn.task; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; SpringBootApplication EnableAsync public class TaskApplication { public static void main(String[] args) { SpringApplication.run(TaskApplication.class, args); } }这里加了EnableAsync后续如果需要在任务中异步回调或异步日志可以直接用Async注解不是必须但提前打开没有坏处。4.2 定义优先级、状态和任务实体先定义优先级枚举。数字越大优先级越高CRITICAL为最高。// 文件路径src/main/java/com/csdn/task/model/TaskPriority.java package com.csdn.task.model; public enum TaskPriority { LOW(1), NORMAL(2), HIGH(3), CRITICAL(4); private final int level; TaskPriority(int level) { this.level level; } public int getLevel() { return level; } }然后定义任务状态枚举。// 文件路径src/main/java/com/csdn/task/model/TaskStatus.java package com.csdn.task.model; public enum TaskStatus { PENDING, RUNNING, SUCCESS, FAILED, RETRYING }接着是任务实体。任务的比较逻辑是优先级系统最核心的地方我们直接让Task实现ComparableTask。// 文件路径src/main/java/com/csdn/task/model/Task.java package com.csdn.task.model; import java.util.concurrent.atomic.AtomicLong; public class Task implements ComparableTask { private static final AtomicLong SEQ new AtomicLong(0); private final String taskId; private final String taskName; private final String payload; private volatile TaskPriority priority; private volatile TaskStatus status; private int retryCount; private long submitTime; private long priorityUpgradeTime; private long sequence; public Task(String taskId, String taskName, String payload, TaskPriority priority) { this.taskId taskId; this.taskName taskName; this.payload payload; this.priority priority; this.status TaskStatus.PENDING; this.submitTime System.currentTimeMillis(); this.priorityUpgradeTime submitTime; this.sequence SEQ.incrementAndGet(); } public String getTaskId() { return taskId; } public String getTaskName() { return taskName; } public String getPayload() { return payload; } public TaskPriority getPriority() { return priority; } public void setPriority(TaskPriority priority) { this.priority priority; } public TaskStatus getStatus() { return status; } public void setStatus(TaskStatus status) { this.status status; } public int getRetryCount() { return retryCount; } public void setRetryCount(int retryCount) { this.retryCount retryCount; } public long getSubmitTime() { return submitTime; } public long getPriorityUpgradeTime() { return priorityUpgradeTime; } public void setPriorityUpgradeTime(long priorityUpgradeTime) { this.priorityUpgradeTime priorityUpgradeTime; } public long getSequence() { return sequence; } Override public int compareTo(Task other) { // 优先比较任务优先级高优在前 int priorityCompare Integer.compare(other.priority.getLevel(), this.priority.getLevel()); if (priorityCompare ! 0) { return priorityCompare; } // 相同优先级遵守先来后到使用自增序号保证公平 return Long.compare(this.sequence, other.sequence); } Override public String toString() { return Task{ taskId taskId \ , taskName taskName \ , priority priority , status status , retryCount retryCount }; } }注意compareTo的写法优先级高的对象应该排在前面所以other.priority.getLevel()与this.priority.getLevel()做比较时要让返回负数表示“当前对象应在前”。同一优先级下使用全局自增序号保证严格公平避免相同优先级的任务出现乱序。4.3 扩展优先级队列支持任务老化PriorityBlockingQueue本身不会动态修改元素优先级。为此我们封装一个PriorityTaskQueue它继承PriorityBlockingQueueTask在每次take()之前先做一次“老化扫描”。// 文件路径src/main/java/com/csdn/task/queue/PriorityTaskQueue.java package com.csdn.task.queue; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.TimeUnit; public class PriorityTaskQueue extends PriorityBlockingQueueTask { private static final long UPGRADE_THRESHOLD_MS 5000L; public PriorityTaskQueue(int capacity) { super(capacity); } Override public Task take() throws InterruptedException { upgradeExpiredTasks(); return super.take(); } Override public Task poll(long timeout, TimeUnit unit) throws InterruptedException { upgradeExpiredTasks(); return super.poll(timeout, unit); } private void upgradeExpiredTasks() { long now System.currentTimeMillis(); for (Task task : this) { if (task.getPriority().getLevel() TaskPriority.CRITICAL.getLevel()) { long waitTime now - task.getPriorityUpgradeTime(); if (waitTime UPGRADE_THRESHOLD_MS) { TaskPriority oldPriority task.getPriority(); TaskPriority newPriority upgradeOnce(oldPriority); if (newPriority ! oldPriority) { // 移除后以新优先级重新插入触发堆重排 boolean removed this.remove(task); if (removed) { task.setPriority(newPriority); task.setPriorityUpgradeTime(now); this.offer(task); } } } } } } private TaskPriority upgradeOnce(TaskPriority priority) { switch (priority) { case LOW: return TaskPriority.NORMAL; case NORMAL: return TaskPriority.HIGH; case HIGH: return TaskPriority.CRITICAL; default: return priority; } } }这个类实现了一个非常朴素的“老化策略”任务在队列里等待超过 5 秒就提升一级优先级再等 5 秒继续提升。这样即使高优任务一直在插队低优任务也会慢慢变成高优最终被消费掉。不过要注意PriorityBlockingQueue的迭代器是弱一致的在take()时扫描整个队列会带来一定开销。本示例的队列规模较小可以接受如果队列中的任务数量达到几十万级不建议每次take()都全量扫描更优做法是使用带时间轮的分桶队列或者定时扫描任务老化。这个我在最佳实践部分会继续说明。4.4 幂等注册表用一个线程安全的 ConcurrentHashMap 记录已成功执行的任务 ID。// 文件路径src/main/java/com/csdn/task/registry/IdempotentTaskRegistry.java package com.csdn.task.registry; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; public class IdempotentTaskRegistry { private final SetString succeededTaskIds ConcurrentHashMap.newKeySet(); public boolean markSucceeded(String taskId) { return succeededTaskIds.add(taskId); } public boolean isSucceeded(String taskId) { return succeededTaskIds.contains(taskId); } public int size() { return succeededTaskIds.size(); } }这里需要注意ConcurrentHashMap.newKeySet()返回的是一个基于 ConcurrentHashMap 的 Set并发安全。add方法返回true表示第一次添加false表示已存在。这正好可以用来判断任务是否已经执行过。4.5 重试策略重试不能无脑重试否则容易把下游服务打挂。我们实现一个简单的指数退避策略每次重试等待时间为baseDelay * (2^n)并且设置一个最大重试次数。// 文件路径src/main/java/com/csdn/task/retry/RetryPolicy.java package com.csdn.task.retry; public class RetryPolicy { private final int maxRetries; private final long baseDelayMillis; public RetryPolicy(int maxRetries, long baseDelayMillis) { this.maxRetries maxRetries; this.baseDelayMillis baseDelayMillis; } public int getMaxRetries() { return maxRetries; } public boolean canRetry(int currentRetryCount) { return currentRetryCount maxRetries; } public long getDelayMillis(int currentRetryCount) { // 指数退避baseDelay * 2^currentRetryCount return baseDelayMillis * (1L currentRetryCount); } }4.6 消费者真正干活的线程消费者是核心执行单元。它从队列中取出任务检查幂等执行业务逻辑失败时按照重试策略决定是直接丢弃还是重新入队。// 文件路径src/main/java/com/csdn/task/consumer/TaskConsumer.java package com.csdn.task.consumer; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import com.csdn.task.model.TaskStatus; import com.csdn.task.queue.PriorityTaskQueue; import com.csdn.task.registry.IdempotentTaskRegistry; import com.csdn.task.retry.RetryPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class TaskConsumer implements Runnable { private static final Logger log LoggerFactory.getLogger(TaskConsumer.class); private final PriorityTaskQueue queue; private final IdempotentTaskRegistry registry; private final RetryPolicy retryPolicy; public TaskConsumer(PriorityTaskQueue queue, IdempotentTaskRegistry registry, RetryPolicy retryPolicy) { this.queue queue; this.registry registry; this.retryPolicy retryPolicy; } Override public void run() { while (!Thread.currentThread().isInterrupted()) { try { Task task queue.take(); // 幂等检查已成功执行过的任务直接跳过 if (registry.isSucceeded(task.getTaskId())) { log.info([幂等跳过] taskId{}, taskName{}, task.getTaskId(), task.getTaskName()); continue; } task.setStatus(TaskStatus.RUNNING); log.info([开始执行] taskId{}, taskName{}, priority{}, task.getTaskId(), task.getTaskName(), task.getPriority()); try { executeBusiness(task); boolean firstMark registry.markSucceeded(task.getTaskId()); task.setStatus(TaskStatus.SUCCESS); log.info([执行成功] taskId{}, taskName{}, firstMark{}, task.getTaskId(), task.getTaskName(), firstMark); } catch (Exception e) { task.setStatus(TaskStatus.FAILED); log.warn([执行失败] taskId{}, taskName{}, error{}, task.getTaskId(), task.getTaskName(), e.getMessage(), e); if (retryPolicy.canRetry(task.getRetryCount())) { long delay retryPolicy.getDelayMillis(task.getRetryCount()); task.setRetryCount(task.getRetryCount() 1); task.setStatus(TaskStatus.RETRYING); log.info([准备重试] taskId{}, 第{}次重试, 延迟{}ms, task.getTaskId(), task.getRetryCount(), delay); Thread.sleep(delay); queue.offer(task); } else { log.error([丢弃任务] taskId{}, taskName{}, 已达到最大重试次数, task.getTaskId(), task.getTaskName()); } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn([消费者中断] {}, e.getMessage()); break; } } } private void executeBusiness(Task task) throws Exception { // 模拟真实业务耗时 Thread.sleep(300L); // 模拟偶发失败统计前先不判断这里直接演示 if (fail-task.equals(task.getPayload())) { throw new RuntimeException(模拟业务异常); } log.info([业务处理] taskName{}, payload{}, task.getTaskName(), task.getPayload()); } }这段代码有几个细节需要说明。第一幂等检查在任务开始执行之前做一次。如果任务在第一次执行时成功但系统没有来得及标记比如标记方法本身出错那么重试时仍会执行一次业务逻辑所以真正的幂等保障还需要业务侧接口支持。调度框架负责大部分情况但不能代替业务幂等。第二任务是重新offer回队列的而不是直接在消费者线程里循环重试。这样做的好处是重试任务会重新参与排队不会因为当前消费者阻塞而影响其他任务坏处是低优任务可能在重试等待后被更高优的任务再次挤后不过因为有老化机制这种延迟是有限的。第三Thread.sleep(delay)是在消费者线程中执行的。如果一个消费者线程在睡觉其他消费者线程不受影响因为线程池通常有多个核心线程。4.7 生产者与任务提交生产者提供了两个方法一个提交普通任务一个提交高优任务。实际项目中生产者往往是业务接口这里用一个独立的TaskProducer类来管理提交逻辑。// 文件路径src/main/java/com/csdn/task/producer/TaskProducer.java package com.csdn.task.producer; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import com.csdn.task.queue.PriorityTaskQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; Component public class TaskProducer { private static final Logger log LoggerFactory.getLogger(TaskProducer.class); private final PriorityTaskQueue queue; public TaskProducer(PriorityTaskQueue queue) { this.queue queue; } public void submitNormalTask(String taskId, String taskName, String payload) { Task task new Task(taskId, taskName, payload, TaskPriority.NORMAL); queue.offer(task); log.info([提交普通任务] taskId{}, taskName{}, taskId, taskName); } public void submitCriticalTask(String taskId, String taskName, String payload) { Task task new Task(taskId, taskName, payload, TaskPriority.CRITICAL); queue.offer(task); log.info([提交高优任务] taskId{}, taskName{}, taskId, taskName); } }4.8 装配线程池与队列Spring 配置类负责将队列、线程池、消费者、生产者组装起来。这一步比较关键直接决定整个调度系统能否正常工作。// 文件路径src/main/java/com/csdn/task/config/TaskSchedulerConfig.java package com.csdn.task.config; import com.csdn.task.consumer.TaskConsumer; import com.csdn.task.producer.TaskProducer; import com.csdn.task.queue.PriorityTaskQueue; import com.csdn.task.registry.IdempotentTaskRegistry; import com.csdn.task.retry.RetryPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; Configuration public class TaskSchedulerConfig { private static final Logger log LoggerFactory.getLogger(TaskSchedulerConfig.class); Bean public PriorityTaskQueue priorityTaskQueue() { return new PriorityTaskQueue(1000); } Bean public IdempotentTaskRegistry idempotentTaskRegistry() { return new IdempotentTaskRegistry(); } Bean public RetryPolicy retryPolicy() { return new RetryPolicy(3, 500L); } Bean(destroyMethod shutdown) public ExecutorService taskExecutor(PriorityTaskQueue queue, IdempotentTaskRegistry registry, RetryPolicy retryPolicy) { int coreThreads 4; int maxThreads 8; long keepAliveSeconds 30L; ThreadFactory threadFactory new ThreadFactory() { private final AtomicInteger counter new AtomicInteger(1); Override public Thread newThread(Runnable r) { Thread thread new Thread(r, task-worker- counter.getAndIncrement()); thread.setDaemon(false); return thread; } }; // 使用有界容量队列防止内存无上限增长 PriorityTaskQueue workQueue queue; ThreadPoolExecutor executor new ThreadPoolExecutor( coreThreads, maxThreads, keepAliveSeconds, TimeUnit.SECONDS, workQueue, threadFactory, new ThreadPoolExecutor.CallerRunsPolicy() ); // 启动消费者线程 for (int i 0; i coreThreads; i) { TaskConsumer consumer new TaskConsumer(queue, registry, retryPolicy); executor.execute(consumer); } log.info([任务调度器] 初始化完成, 核心线程数{}, 最大线程数{}, coreThreads, maxThreads); return executor; } Bean public TaskProducer taskProducer(PriorityTaskQueue queue) { return new TaskProducer(queue); } }这里有一个比较容易踩坑的点ThreadPoolExecutor.execute()本身是用于提交任务的但我们在初始化时直接使用executor.execute(consumer)来启动消费者的循环任务。这会占用核心线程并且这些线程会一直阻塞在queue.take()上属于预期行为不需要担心。另一个点是destroyMethod shutdown这样在 Spring 容器关闭时Spring 会自动调用线程池的shutdown()方法避免了线程资源泄漏。4.9 编写测试接口为了直观演示效果我写一个简单的 HTTP 接口。接口提供两个端点一个提交普通任务一个提交高优任务。// 文件路径src/main/java/com/csdn/task/controller/TaskController.java package com.csdn.task.controller; import com.csdn.task.producer.TaskProducer; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; RestController RequestMapping(/task) public class TaskController { private final TaskProducer taskProducer; public TaskController(TaskProducer taskProducer) { this.taskProducer taskProducer; } GetMapping(/submit) public String submit(RequestParam(type) String type, RequestParam(taskId) String taskId, RequestParam(taskName) String taskName, RequestParam(value payload, defaultValue ) String payload) { if (critical.equalsIgnoreCase(type)) { taskProducer.submitCriticalTask(taskId, taskName, payload); return critical task submitted: taskId; } taskProducer.submitNormalTask(taskId, taskName, payload); return normal task submitted: taskId; } }4.10 运行与验证启动TaskApplication然后在浏览器或 curl 中访问接口。先连续提交 20 个普通任务for i in $(seq 1 20); do curl http://localhost:8080/task/submit?typenormaltaskIdnormal-$itaskName批量任务$ipayloadnormal done再提交一个高优任务curl http://localhost:8080/task/submit?typecriticaltaskIdcritical-1taskName紧急订单处理payloadimportant观察日志。你会在高优任务提交后很快看到[开始执行]日志说明高优任务插队成功。与此同时一些普通任务的[开始执行]日志也会出现说明它们没有被完全饿死。下面是一段可能的日志输出顺序以实际运行环境为准[提交普通任务] taskIdnormal-1, taskName批量任务1 [提交普通任务] taskIdnormal-2, taskName批量任务2 ... [提交高优任务] taskIdcritical-1, taskName紧急订单处理 [开始执行] taskIdcritical-1, taskName紧急订单处理, priorityCRITICAL [开始执行] taskIdnormal-1, taskName批量任务1, priorityNORMAL如果想验证幂等可以再次提交相同的任务 IDcurl http://localhost:8080/task/submit?typenormaltaskIdnormal-1taskName重复任务payloadnormal日志中会出现[幂等跳过]说明相同的任务 ID 不会重复执行。如果想要验证重试可以在提交高优任务时把 payload 设为fail-taskcurl http://localhost:8080/task/submit?typecriticaltaskIdfail-1taskName异常任务payloadfail-task日志中会出现[执行失败]、[准备重试]重试 3 次后进入[丢弃任务]。5. 常见问题与排查思路任务调度系统在开发和使用过程中问题往往集中在“队列堆积”“优先级不生效”“任务丢失”“重复执行”这几个方向上。下面整理一张问题排查表方便你遇到问题时快速定位。问题现象常见原因解决思路高优任务迟迟不执行生产者提交后队列未生效优先级比较器写反消费者线程数过少检查compareTo方向确认提交的任务确实调用了高优方法观察线程池活跃线程数低优任务长时间不执行高优任务持续挤占队列头部增加任务老化机制拆分为多个独立队列按权重分配消费名额相同优先级任务乱序compareTo只比较优先级没有保持先进先出在比较逻辑中加入自增序号先提交的任务排前面任务执行成功但日志显示重试执行成功后标记幂等失败业务方法中捕获了异常确认markSucceeded是否在executeBusiness之后立即调用检查业务方法是否吞掉了异常线程池满导致任务丢失默认拒绝策略抛异常且无人处理使用CallerRunsPolicy或捕获拒绝异常重新投递重复提交同一任务导致重复执行缺少幂等判定任务 ID 生成规则不稳定引入幂等注册表保证每种任务有稳定的业务唯一键消费者停止消费线程被中断任务执行方法抛Error而非Exception在catch (Exception)基础上增加catch (Throwable)兜底检查线程是否被误杀这里单独说明一个非常隐蔽的问题compareTo的方向。PriorityBlockingQueue是小顶堆实现它认为“最小的”元素最先出队。如果你希望优先级高的先出队那么高优对象的compareTo应该返回负数。初学者很容易把这个方向写反导致高优任务反而最后执行。排查时可以打印队列中的任务顺序先确认比较结果是否符合预期。另一个常见问题是消费者线程数量与核心线程数不等。上面配置里我启动消费者的方式是executor.execute(consumer)这会把核心线程全部占满。如果你手动提交了大量外部任务会发现外部任务没有空闲线程可以执行因为核心线程都在循环take()任务。这里其实有两种设计选择一是消费者即线程池的工作线程线程池不再对外接收其他任务二是把消费线程单独再建一个线程池而当前线程池只负责业务执行。实际项目中需要根据你的业务形态选择不能混用。6. 最佳实践与工程建议6.1 不要在大流量场景使用全量老化扫描示例中的老化方案在队列元素很少时没有问题但当队列积压达到几十万条时每次take()遍历全队列会消耗大量 CPU反而拖慢调度性能。工程上建议改用以下方案之一独立老化线程每隔固定时间扫描一次队列而不是每次取任务时扫描。分层队列按优先级拆成多个普通队列调度线程按照权重从高优到低优获取任务并记录低优队列等待时间。时间轮分桶按等待时间放入不同桶过期后统一升级。如果你的系统任务量超过单机处理能力优先考虑引入消息队列不要让单机队列承担过重的堆积职责。6.2 线程池参数需要根据任务特征调优线程数和队列容量的设置取决于你的任务属于 CPU 密集型还是 IO 密集型。CPU 密集型任务线程数建议接近 CPU 核数。IO 密集型任务线程数可以适当调大比如CPU核数 * 2或更高。队列容量必须是有界的。无界队列在流量突发时会导致内存暴涨进而引发 OOM。建议在你的压测环境里分别调整核心线程数和队列大小观察吞吐量、平均延迟、拒绝次数三个指标。不要仅仅照搬别人的配置。6.3 幂等设计是任务系统的底线本文使用的是本地内存Set只能解决单机内重复提交。生产环境的任务系统一定是多实例部署所以必须使用分布式幂等存储。推荐方案Redis使用SETNX命令任务 ID 作为 Key首次设置成功才允许执行。数据库唯一索引在任务表中给task_id加唯一约束插入冲突则说明已处理。消息队列自带去重比如 RocketMQ 的消息幂等消费需要业务方配合但框架不会自动去重。同时必须提醒一点调度框架的幂等只能防止“同一任务提交两次”不能消除业务逻辑本身的重复副作用。比如发送短信、扣减库存、转账等操作一定要在业务代码里再次做幂等校验。6.4 失败任务要区分“可重试”和“不可重试”并非所有异常都值得重试。网络超时、数据库连接短暂不可用、下游限流这些属于可重试参数错误、数据非法、业务状态不允许操作这些属于不可重试。如果对所有异常都重试不仅浪费资源还可能放大问题。建议在业务执行方法中封装一个异常分类public class BizException extends RuntimeException { private final boolean retryable; public BizException(String message, boolean retryable) { super(message); this.retryable retryable; } public boolean isRetryable() { return retryable; } }消费者在捕获异常时判断retryable字段只有可重试异常才进入重试逻辑。这样也可以避免无效重试把任务打到最大次数。6.5 增加可观测性任务调度系统属于异步链路一旦出问题很难直接感知。必须在代码中埋入足够的监控指标队列当前长度。每个优先级的任务数量。任务平均等待时间。任务执行成功率、失败率。每条任务的执行耗时分布。重试次数分布。在 Spring Boot 中可以通过Actuator暴露指标也可以使用Micrometer与 Prometheus 对接。如果没有现成的监控系统至少要在日志中打印关键指标方便线上排查。6.6 从单机队列到分布式消息队列的平滑演进当项目规模变大单机队列会逐渐成为瓶颈。建议演进路径如下阶段一单体应用任务量不大使用本文方案。阶段二任务量增长拆出独立的任务处理服务使用消息队列解耦生产者和消费者。阶段三需要严格全局优先级将高优任务单独发往一个高优先级的 Topic低优任务发往另一个 Topic消费者各自维护消费线程池并设置权重。在阶段三中不要试图在 Kafka 这类消息系统里实现“队列内动态插队”因为 Kafka 的分区顺序模型不支持这种操作。正确的做法是物理拆分 Topic用多个消费者组或消费优先级来解决。7. 总结与下一步回到刚开始那句话“苍生我要救钟离权我也要救。”在现实系统中这从来不是一个单选题。通过优先级队列 任务老化 重试 幂等这个组合方案我们可以让高优任务及时被处理同时低优任务也不会被永久饿死。本文通过一个完整的 Spring Boot 示例讲解了如何用PriorityBlockingQueue 自定义compareTo实现高优任务插队。如何用任务老化机制避免低优任务饿死。如何设计任务状态机让每个任务都有清晰的状态流转。如何在失败时用指数退避策略重试避免打垮下游。如何用任务 ID 做幂等防止重复执行。遇到问题时的排查顺序和常见坑点。如果这是你第一次接触任务调度下一步可以继续学习ThreadPoolExecutor的原理、ScheduledExecutorService的定时调度然后逐步尝试把任务调度扩展到 RocketMQ 或 RabbitMQ 的分布式场景。在实际项目中优先关注三件事线程池参数是否经过压测、幂等是否覆盖所有关键任务、监控告警是否在任务积压时第一时间通知你。把这三件事做好任务系统基本就能稳定运行。如果这篇文章对你有帮助建议收藏备用下次设计任务调度系统时可以直接对照本文的代码和排查表来落地。
“苍生我要救钟离权我也要救”——这句话初看像一句仙侠剧情台词但放在后端系统里它其实就是我们经常遇到的两难局面一边是海量普通任务在不断积压一边是某个核心高优任务必须立即处理。很多同学在处理这种需求时容易走极端要么只保高优任务导致普通任务饿死要么把所有任务一视同仁结果核心任务被长尾任务拖住。本文就围绕这个“两边都要保”的场景拆解一套优先级任务调度方案包含核心原理、完整可运行代码、常见排队问题和工程落地建议。无论你是刚接触并发编程的后端新手还是正在设计任务系统的中台工程师都可以把这段内容当作一份可直接复用的实操笔记。1. 背景与核心概念1.1 从“救苍生”到“任务调度”的映射先把这个标题翻译成技术语言。所谓“救苍生”对应的是业务系统中大量常规任务日志上报、短信通知、数据同步、批量导出、定时对账等。这类任务数量大、单条耗时短、对实时性要求不极端但它们占据了系统绝大部分资源。所谓“救钟离权”则对应一个或少数几个高优核心任务比如用户支付后的订单处理、风控拦截通知、紧急配置刷新、资金动账请求。这类任务数量少但延迟高了就要出事故。问题在于如果只“救苍生”按照先来先服务的方式排队那么一个高优订单处理任务可能排在上千个日志任务后面等轮到它时业务已经超时如果只“救钟离权”来了高优任务就无限插队普通任务可能在极端流量下一直得不到执行最终堆积成雪崩。所以这篇文章要解决的就是如何在同一套任务处理系统里既保证高优任务的低延迟又保证普通任务最终都能被执行不出现饿死。1.2 任务调度系统的基本组成任何任务调度系统不管简单还是复杂都离不开四个核心部分任务输入外部请求把任务提交进系统包括任务ID、优先级、类型、业务数据等。任务队列暂存待执行任务并根据某种策略决定下一个执行谁。任务执行单元真正运行任务的线程或进程。任务生命周期管理从提交、等待、执行、成功、失败、重试到最终结束的完整状态流转。在此基础上优先级任务调度就是在“任务队列”和“执行单元”之间增加一套取舍规则高优任务优先被拿走低优任务不因高优任务持续到达而永久等待。1.3 单机方案与分布式方案的边界在正式写代码之前必须先明确一个边界本文实现的调度器是一个单机进程内的方案适合单体应用、服务内异步任务、后台线程池任务处理等场景。它不直接解决分布式多节点下的全局优先级调度问题。如果你的系统已经引入了消息队列比如 RocketMQ、RabbitMQ、Kafka并且需要在多个消费节点之间做优先级调度那么单机的 PriorityBlockingQueue 就不够用了。截至当前主流实践分布式优先级调度一般会采用“分区主题 按优先级拆分Topic”或“同一Topic内结合消息过滤/延迟消息”等方式实现这部分在最后的最佳实践里会给出建议。不区分单机和分布式就直接讲队列是很多教程容易误导人的地方。我们先稳住单机场景把原理吃透再去拓展分布式。2. 环境准备与版本说明2.1 运行环境本文示例使用 Java 语言和 Spring Boot 框架。之所以选这个组合是因为 Spring Boot 的自动装配和参数配置能让我们把精力集中在调度逻辑上而不是浪费在容器初始化上。你需要准备的基础环境如下JDK 8 及以上版本示例代码用到了 Lambda、Stream 等特性JDK 8 即可。Maven 3.6 以上用于管理依赖。Spring Boot 2.x 或 3.x 均可本文示例以 Spring Boot 2.x 常见写法为主如果你用的是 Spring Boot 3.x依赖坐标中的javax需要替换为jakarta但核心调度代码基本不用改。IDE 推荐 IntelliJ IDEA命令行演示不强制依赖 IDE。版本需要根据你的项目实际情况调整不必照搬某个具体小版本。优先确认你本机已安装 JDK 且java -version能正常输出。2.2 示例项目结构本文实战案例的代码结构如下priority-task-scheduler/ ├── pom.xml └── src/main/ ├── java/com/csdn/task/ │ ├── TaskApplication.java │ ├── model/ │ │ ├── TaskPriority.java │ │ ├── TaskStatus.java │ │ └── Task.java │ ├── queue/ │ │ └── PriorityTaskQueue.java │ ├── producer/ │ │ └── TaskProducer.java │ ├── consumer/ │ │ └── TaskConsumer.java │ ├── retry/ │ │ └── RetryPolicy.java │ └── registry/ │ └── IdempotentTaskRegistry.java └── resources/ └── application.yml这个结构很清晰模型层放任务实体、状态枚举和优先级枚举队列层封装优先级队列生产者负责接收外部任务消费者负责执行重试策略和幂等注册表分别解决失败重试与重复执行问题。2.3 依赖配置在pom.xml中添加 Spring Boot Web 依赖就够了这里用 Web 只是为了提供 HTTP 入口来测试任务提交生产环境中完全可以用业务接口替代。?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.csdn/groupId artifactIdpriority-task-scheduler/artifactId version1.0.0/version packagingjar/packaging parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version relativePath/ /parent properties java.version1.8/java.version /properties dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project如果你的项目已经存在不需要刻意改成这个结构理解每个类的职责后直接迁移到你的工程里即可。3. 核心原理拆解3.1 优先级队列 PriorityBlockingQueueJava 并发包中提供了一个非常合适的数据结构PriorityBlockingQueue。它是一个基于优先级堆的无界阻塞队列线程安全且出队时会按照元素的自然顺序或自定义比较器返回最高优先级的元素。也就是说多个消费者线程同时从队列中take()任务时永远不会拿到“次优”的任务而是拿到当前队列里优先级最高的那个。这一特性天然适合“救钟离权”。但这里有一个关键设计点任务的“优先级”如何定义通常有两种做法用整数表示数字越小优先级越高。用枚举表示通过compareTo方法实现排序。第二种更可读也更容易在业务代码里使用。下面我会采用枚举方式。3.2 高优插队与 low 优先级的饿死问题优先级队列带来的第一个副作用就是“低优先级任务饿死”。如果高优任务源源不断提交进来低优任务可能永远轮不到执行。这在业务上表现为普通日志任务越积越多高优任务持续插队最终普通任务延迟以小时级增长。解决饿死问题业界常用的思路有以下几种任务老化任务在队列中等待超过一定时间后动态提升优先级。权重轮转按优先级权重分配执行名额比如每执行 10 个高优任务至少执行 1 个低优任务。独立队列隔离完全把高低优任务拆到不同线程池避免互相抢占。本文示例采用“任务老化”因为它实现成本低且能直观演示优先级变化过程。3.3 线程池的选择有优先级队列还不够还需要一个线程池来消费任务。Java 的ThreadPoolExecutor支持传入一个BlockingQueue作为工作队列所以我们只需要把PriorityTaskQueue传给ThreadPoolExecutor即可。线程池的核心参数需要认真说明corePoolSize核心线程数常驻线程数量。maximumPoolSize最大线程数任务高峰期最多创建的线程数量。keepAliveTime非核心线程空闲存活时间。unit存活时间的单位。workQueue工作队列必须使用我们的优先级队列。threadFactory线程工厂用来给线程取有意义的名字方便排查日志。handler拒绝策略当线程池和队列都满时触发。拒绝策略是一个非常值得注意的点。默认的AbortPolicy会在队列满时直接抛异常导致任务丢失。对于任务调度系统我更推荐CallerRunsPolicy它会让提交任务的线程自己去执行任务虽然会阻塞生产者但至少任务不会丢。不过CallerRunsPolicy在优先级队列场景下有一个副作用被拒绝回调用线程执行的任务可能会绕过优先级顺序。所以如果你的场景对优先级极其敏感建议不用CallerRunsPolicy而是配合重试机制将失败任务重新投递。3.4 任务状态机一个任务从提交到结束至少要经历以下状态PENDING → RUNNING → SUCCESS ↓ FAILED → RETRYING → RUNNINGPENDING已入队等待执行。RUNNING已从队列取出正在执行。SUCCESS执行成功流程结束。FAILED执行失败进入失败处理流程。RETRYING满足重试条件准备重新入队。为什么要设计这么一套状态机因为只有状态明确才能回答三个关键问题任务是否执行过、执行到一半失败怎么办、重复投递是否安全。很多线上事故都是因为状态不清晰重复消费或失败后静默丢弃。3.5 幂等保障为什么必须考虑任务调度系统在处理“任务丢失”时第一反应往往是“重试”。但重试可能造成重复执行。比如一个订单通知任务第一次执行时网络超时实际上服务端已经收到并处理了重试又发一次用户就收到两条一模一样的通知。解决的办法有两个一是让业务处理逻辑本身支持幂等同一任务执行多次结果一致二是在调度框架层面做去重。框架层面的去重一般依赖任务 ID。本文示例会用ConcurrentHashMap维护一个已执行成功的任务 ID 集合相同 ID 的任务不允许重复执行。这只是单体场景演示分布式场景请使用 Redis 等外部存储做分布式幂等最佳实践部分会提到。4. 完整实战案例优先保障 批量兜底的任务调度器现在进入核心实战。我们的目标是实现这样一个效果普通任务和服务可以随便提交不会被拒绝。高优任务提交后能在 1 秒内被线程取出执行。普通任务即使前面堆了很多高优任务也不会饿死而是等待超过一定时间后自动升级优先级。失败任务最多重试 3 次每次退避时间递增。同一个任务 ID 不会被重复执行。4.1 创建项目结构和启动类先用 Maven 创建工程或者直接在 IDEA 里 Spring Initializr 创建。完成后先写启动类。// 文件路径src/main/java/com/csdn/task/TaskApplication.java package com.csdn.task; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; SpringBootApplication EnableAsync public class TaskApplication { public static void main(String[] args) { SpringApplication.run(TaskApplication.class, args); } }这里加了EnableAsync后续如果需要在任务中异步回调或异步日志可以直接用Async注解不是必须但提前打开没有坏处。4.2 定义优先级、状态和任务实体先定义优先级枚举。数字越大优先级越高CRITICAL为最高。// 文件路径src/main/java/com/csdn/task/model/TaskPriority.java package com.csdn.task.model; public enum TaskPriority { LOW(1), NORMAL(2), HIGH(3), CRITICAL(4); private final int level; TaskPriority(int level) { this.level level; } public int getLevel() { return level; } }然后定义任务状态枚举。// 文件路径src/main/java/com/csdn/task/model/TaskStatus.java package com.csdn.task.model; public enum TaskStatus { PENDING, RUNNING, SUCCESS, FAILED, RETRYING }接着是任务实体。任务的比较逻辑是优先级系统最核心的地方我们直接让Task实现ComparableTask。// 文件路径src/main/java/com/csdn/task/model/Task.java package com.csdn.task.model; import java.util.concurrent.atomic.AtomicLong; public class Task implements ComparableTask { private static final AtomicLong SEQ new AtomicLong(0); private final String taskId; private final String taskName; private final String payload; private volatile TaskPriority priority; private volatile TaskStatus status; private int retryCount; private long submitTime; private long priorityUpgradeTime; private long sequence; public Task(String taskId, String taskName, String payload, TaskPriority priority) { this.taskId taskId; this.taskName taskName; this.payload payload; this.priority priority; this.status TaskStatus.PENDING; this.submitTime System.currentTimeMillis(); this.priorityUpgradeTime submitTime; this.sequence SEQ.incrementAndGet(); } public String getTaskId() { return taskId; } public String getTaskName() { return taskName; } public String getPayload() { return payload; } public TaskPriority getPriority() { return priority; } public void setPriority(TaskPriority priority) { this.priority priority; } public TaskStatus getStatus() { return status; } public void setStatus(TaskStatus status) { this.status status; } public int getRetryCount() { return retryCount; } public void setRetryCount(int retryCount) { this.retryCount retryCount; } public long getSubmitTime() { return submitTime; } public long getPriorityUpgradeTime() { return priorityUpgradeTime; } public void setPriorityUpgradeTime(long priorityUpgradeTime) { this.priorityUpgradeTime priorityUpgradeTime; } public long getSequence() { return sequence; } Override public int compareTo(Task other) { // 优先比较任务优先级高优在前 int priorityCompare Integer.compare(other.priority.getLevel(), this.priority.getLevel()); if (priorityCompare ! 0) { return priorityCompare; } // 相同优先级遵守先来后到使用自增序号保证公平 return Long.compare(this.sequence, other.sequence); } Override public String toString() { return Task{ taskId taskId \ , taskName taskName \ , priority priority , status status , retryCount retryCount }; } }注意compareTo的写法优先级高的对象应该排在前面所以other.priority.getLevel()与this.priority.getLevel()做比较时要让返回负数表示“当前对象应在前”。同一优先级下使用全局自增序号保证严格公平避免相同优先级的任务出现乱序。4.3 扩展优先级队列支持任务老化PriorityBlockingQueue本身不会动态修改元素优先级。为此我们封装一个PriorityTaskQueue它继承PriorityBlockingQueueTask在每次take()之前先做一次“老化扫描”。// 文件路径src/main/java/com/csdn/task/queue/PriorityTaskQueue.java package com.csdn.task.queue; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.TimeUnit; public class PriorityTaskQueue extends PriorityBlockingQueueTask { private static final long UPGRADE_THRESHOLD_MS 5000L; public PriorityTaskQueue(int capacity) { super(capacity); } Override public Task take() throws InterruptedException { upgradeExpiredTasks(); return super.take(); } Override public Task poll(long timeout, TimeUnit unit) throws InterruptedException { upgradeExpiredTasks(); return super.poll(timeout, unit); } private void upgradeExpiredTasks() { long now System.currentTimeMillis(); for (Task task : this) { if (task.getPriority().getLevel() TaskPriority.CRITICAL.getLevel()) { long waitTime now - task.getPriorityUpgradeTime(); if (waitTime UPGRADE_THRESHOLD_MS) { TaskPriority oldPriority task.getPriority(); TaskPriority newPriority upgradeOnce(oldPriority); if (newPriority ! oldPriority) { // 移除后以新优先级重新插入触发堆重排 boolean removed this.remove(task); if (removed) { task.setPriority(newPriority); task.setPriorityUpgradeTime(now); this.offer(task); } } } } } } private TaskPriority upgradeOnce(TaskPriority priority) { switch (priority) { case LOW: return TaskPriority.NORMAL; case NORMAL: return TaskPriority.HIGH; case HIGH: return TaskPriority.CRITICAL; default: return priority; } } }这个类实现了一个非常朴素的“老化策略”任务在队列里等待超过 5 秒就提升一级优先级再等 5 秒继续提升。这样即使高优任务一直在插队低优任务也会慢慢变成高优最终被消费掉。不过要注意PriorityBlockingQueue的迭代器是弱一致的在take()时扫描整个队列会带来一定开销。本示例的队列规模较小可以接受如果队列中的任务数量达到几十万级不建议每次take()都全量扫描更优做法是使用带时间轮的分桶队列或者定时扫描任务老化。这个我在最佳实践部分会继续说明。4.4 幂等注册表用一个线程安全的 ConcurrentHashMap 记录已成功执行的任务 ID。// 文件路径src/main/java/com/csdn/task/registry/IdempotentTaskRegistry.java package com.csdn.task.registry; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; public class IdempotentTaskRegistry { private final SetString succeededTaskIds ConcurrentHashMap.newKeySet(); public boolean markSucceeded(String taskId) { return succeededTaskIds.add(taskId); } public boolean isSucceeded(String taskId) { return succeededTaskIds.contains(taskId); } public int size() { return succeededTaskIds.size(); } }这里需要注意ConcurrentHashMap.newKeySet()返回的是一个基于 ConcurrentHashMap 的 Set并发安全。add方法返回true表示第一次添加false表示已存在。这正好可以用来判断任务是否已经执行过。4.5 重试策略重试不能无脑重试否则容易把下游服务打挂。我们实现一个简单的指数退避策略每次重试等待时间为baseDelay * (2^n)并且设置一个最大重试次数。// 文件路径src/main/java/com/csdn/task/retry/RetryPolicy.java package com.csdn.task.retry; public class RetryPolicy { private final int maxRetries; private final long baseDelayMillis; public RetryPolicy(int maxRetries, long baseDelayMillis) { this.maxRetries maxRetries; this.baseDelayMillis baseDelayMillis; } public int getMaxRetries() { return maxRetries; } public boolean canRetry(int currentRetryCount) { return currentRetryCount maxRetries; } public long getDelayMillis(int currentRetryCount) { // 指数退避baseDelay * 2^currentRetryCount return baseDelayMillis * (1L currentRetryCount); } }4.6 消费者真正干活的线程消费者是核心执行单元。它从队列中取出任务检查幂等执行业务逻辑失败时按照重试策略决定是直接丢弃还是重新入队。// 文件路径src/main/java/com/csdn/task/consumer/TaskConsumer.java package com.csdn.task.consumer; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import com.csdn.task.model.TaskStatus; import com.csdn.task.queue.PriorityTaskQueue; import com.csdn.task.registry.IdempotentTaskRegistry; import com.csdn.task.retry.RetryPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class TaskConsumer implements Runnable { private static final Logger log LoggerFactory.getLogger(TaskConsumer.class); private final PriorityTaskQueue queue; private final IdempotentTaskRegistry registry; private final RetryPolicy retryPolicy; public TaskConsumer(PriorityTaskQueue queue, IdempotentTaskRegistry registry, RetryPolicy retryPolicy) { this.queue queue; this.registry registry; this.retryPolicy retryPolicy; } Override public void run() { while (!Thread.currentThread().isInterrupted()) { try { Task task queue.take(); // 幂等检查已成功执行过的任务直接跳过 if (registry.isSucceeded(task.getTaskId())) { log.info([幂等跳过] taskId{}, taskName{}, task.getTaskId(), task.getTaskName()); continue; } task.setStatus(TaskStatus.RUNNING); log.info([开始执行] taskId{}, taskName{}, priority{}, task.getTaskId(), task.getTaskName(), task.getPriority()); try { executeBusiness(task); boolean firstMark registry.markSucceeded(task.getTaskId()); task.setStatus(TaskStatus.SUCCESS); log.info([执行成功] taskId{}, taskName{}, firstMark{}, task.getTaskId(), task.getTaskName(), firstMark); } catch (Exception e) { task.setStatus(TaskStatus.FAILED); log.warn([执行失败] taskId{}, taskName{}, error{}, task.getTaskId(), task.getTaskName(), e.getMessage(), e); if (retryPolicy.canRetry(task.getRetryCount())) { long delay retryPolicy.getDelayMillis(task.getRetryCount()); task.setRetryCount(task.getRetryCount() 1); task.setStatus(TaskStatus.RETRYING); log.info([准备重试] taskId{}, 第{}次重试, 延迟{}ms, task.getTaskId(), task.getRetryCount(), delay); Thread.sleep(delay); queue.offer(task); } else { log.error([丢弃任务] taskId{}, taskName{}, 已达到最大重试次数, task.getTaskId(), task.getTaskName()); } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn([消费者中断] {}, e.getMessage()); break; } } } private void executeBusiness(Task task) throws Exception { // 模拟真实业务耗时 Thread.sleep(300L); // 模拟偶发失败统计前先不判断这里直接演示 if (fail-task.equals(task.getPayload())) { throw new RuntimeException(模拟业务异常); } log.info([业务处理] taskName{}, payload{}, task.getTaskName(), task.getPayload()); } }这段代码有几个细节需要说明。第一幂等检查在任务开始执行之前做一次。如果任务在第一次执行时成功但系统没有来得及标记比如标记方法本身出错那么重试时仍会执行一次业务逻辑所以真正的幂等保障还需要业务侧接口支持。调度框架负责大部分情况但不能代替业务幂等。第二任务是重新offer回队列的而不是直接在消费者线程里循环重试。这样做的好处是重试任务会重新参与排队不会因为当前消费者阻塞而影响其他任务坏处是低优任务可能在重试等待后被更高优的任务再次挤后不过因为有老化机制这种延迟是有限的。第三Thread.sleep(delay)是在消费者线程中执行的。如果一个消费者线程在睡觉其他消费者线程不受影响因为线程池通常有多个核心线程。4.7 生产者与任务提交生产者提供了两个方法一个提交普通任务一个提交高优任务。实际项目中生产者往往是业务接口这里用一个独立的TaskProducer类来管理提交逻辑。// 文件路径src/main/java/com/csdn/task/producer/TaskProducer.java package com.csdn.task.producer; import com.csdn.task.model.Task; import com.csdn.task.model.TaskPriority; import com.csdn.task.queue.PriorityTaskQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; Component public class TaskProducer { private static final Logger log LoggerFactory.getLogger(TaskProducer.class); private final PriorityTaskQueue queue; public TaskProducer(PriorityTaskQueue queue) { this.queue queue; } public void submitNormalTask(String taskId, String taskName, String payload) { Task task new Task(taskId, taskName, payload, TaskPriority.NORMAL); queue.offer(task); log.info([提交普通任务] taskId{}, taskName{}, taskId, taskName); } public void submitCriticalTask(String taskId, String taskName, String payload) { Task task new Task(taskId, taskName, payload, TaskPriority.CRITICAL); queue.offer(task); log.info([提交高优任务] taskId{}, taskName{}, taskId, taskName); } }4.8 装配线程池与队列Spring 配置类负责将队列、线程池、消费者、生产者组装起来。这一步比较关键直接决定整个调度系统能否正常工作。// 文件路径src/main/java/com/csdn/task/config/TaskSchedulerConfig.java package com.csdn.task.config; import com.csdn.task.consumer.TaskConsumer; import com.csdn.task.producer.TaskProducer; import com.csdn.task.queue.PriorityTaskQueue; import com.csdn.task.registry.IdempotentTaskRegistry; import com.csdn.task.retry.RetryPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; Configuration public class TaskSchedulerConfig { private static final Logger log LoggerFactory.getLogger(TaskSchedulerConfig.class); Bean public PriorityTaskQueue priorityTaskQueue() { return new PriorityTaskQueue(1000); } Bean public IdempotentTaskRegistry idempotentTaskRegistry() { return new IdempotentTaskRegistry(); } Bean public RetryPolicy retryPolicy() { return new RetryPolicy(3, 500L); } Bean(destroyMethod shutdown) public ExecutorService taskExecutor(PriorityTaskQueue queue, IdempotentTaskRegistry registry, RetryPolicy retryPolicy) { int coreThreads 4; int maxThreads 8; long keepAliveSeconds 30L; ThreadFactory threadFactory new ThreadFactory() { private final AtomicInteger counter new AtomicInteger(1); Override public Thread newThread(Runnable r) { Thread thread new Thread(r, task-worker- counter.getAndIncrement()); thread.setDaemon(false); return thread; } }; // 使用有界容量队列防止内存无上限增长 PriorityTaskQueue workQueue queue; ThreadPoolExecutor executor new ThreadPoolExecutor( coreThreads, maxThreads, keepAliveSeconds, TimeUnit.SECONDS, workQueue, threadFactory, new ThreadPoolExecutor.CallerRunsPolicy() ); // 启动消费者线程 for (int i 0; i coreThreads; i) { TaskConsumer consumer new TaskConsumer(queue, registry, retryPolicy); executor.execute(consumer); } log.info([任务调度器] 初始化完成, 核心线程数{}, 最大线程数{}, coreThreads, maxThreads); return executor; } Bean public TaskProducer taskProducer(PriorityTaskQueue queue) { return new TaskProducer(queue); } }这里有一个比较容易踩坑的点ThreadPoolExecutor.execute()本身是用于提交任务的但我们在初始化时直接使用executor.execute(consumer)来启动消费者的循环任务。这会占用核心线程并且这些线程会一直阻塞在queue.take()上属于预期行为不需要担心。另一个点是destroyMethod shutdown这样在 Spring 容器关闭时Spring 会自动调用线程池的shutdown()方法避免了线程资源泄漏。4.9 编写测试接口为了直观演示效果我写一个简单的 HTTP 接口。接口提供两个端点一个提交普通任务一个提交高优任务。// 文件路径src/main/java/com/csdn/task/controller/TaskController.java package com.csdn.task.controller; import com.csdn.task.producer.TaskProducer; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; RestController RequestMapping(/task) public class TaskController { private final TaskProducer taskProducer; public TaskController(TaskProducer taskProducer) { this.taskProducer taskProducer; } GetMapping(/submit) public String submit(RequestParam(type) String type, RequestParam(taskId) String taskId, RequestParam(taskName) String taskName, RequestParam(value payload, defaultValue ) String payload) { if (critical.equalsIgnoreCase(type)) { taskProducer.submitCriticalTask(taskId, taskName, payload); return critical task submitted: taskId; } taskProducer.submitNormalTask(taskId, taskName, payload); return normal task submitted: taskId; } }4.10 运行与验证启动TaskApplication然后在浏览器或 curl 中访问接口。先连续提交 20 个普通任务for i in $(seq 1 20); do curl http://localhost:8080/task/submit?typenormaltaskIdnormal-$itaskName批量任务$ipayloadnormal done再提交一个高优任务curl http://localhost:8080/task/submit?typecriticaltaskIdcritical-1taskName紧急订单处理payloadimportant观察日志。你会在高优任务提交后很快看到[开始执行]日志说明高优任务插队成功。与此同时一些普通任务的[开始执行]日志也会出现说明它们没有被完全饿死。下面是一段可能的日志输出顺序以实际运行环境为准[提交普通任务] taskIdnormal-1, taskName批量任务1 [提交普通任务] taskIdnormal-2, taskName批量任务2 ... [提交高优任务] taskIdcritical-1, taskName紧急订单处理 [开始执行] taskIdcritical-1, taskName紧急订单处理, priorityCRITICAL [开始执行] taskIdnormal-1, taskName批量任务1, priorityNORMAL如果想验证幂等可以再次提交相同的任务 IDcurl http://localhost:8080/task/submit?typenormaltaskIdnormal-1taskName重复任务payloadnormal日志中会出现[幂等跳过]说明相同的任务 ID 不会重复执行。如果想要验证重试可以在提交高优任务时把 payload 设为fail-taskcurl http://localhost:8080/task/submit?typecriticaltaskIdfail-1taskName异常任务payloadfail-task日志中会出现[执行失败]、[准备重试]重试 3 次后进入[丢弃任务]。5. 常见问题与排查思路任务调度系统在开发和使用过程中问题往往集中在“队列堆积”“优先级不生效”“任务丢失”“重复执行”这几个方向上。下面整理一张问题排查表方便你遇到问题时快速定位。问题现象常见原因解决思路高优任务迟迟不执行生产者提交后队列未生效优先级比较器写反消费者线程数过少检查compareTo方向确认提交的任务确实调用了高优方法观察线程池活跃线程数低优任务长时间不执行高优任务持续挤占队列头部增加任务老化机制拆分为多个独立队列按权重分配消费名额相同优先级任务乱序compareTo只比较优先级没有保持先进先出在比较逻辑中加入自增序号先提交的任务排前面任务执行成功但日志显示重试执行成功后标记幂等失败业务方法中捕获了异常确认markSucceeded是否在executeBusiness之后立即调用检查业务方法是否吞掉了异常线程池满导致任务丢失默认拒绝策略抛异常且无人处理使用CallerRunsPolicy或捕获拒绝异常重新投递重复提交同一任务导致重复执行缺少幂等判定任务 ID 生成规则不稳定引入幂等注册表保证每种任务有稳定的业务唯一键消费者停止消费线程被中断任务执行方法抛Error而非Exception在catch (Exception)基础上增加catch (Throwable)兜底检查线程是否被误杀这里单独说明一个非常隐蔽的问题compareTo的方向。PriorityBlockingQueue是小顶堆实现它认为“最小的”元素最先出队。如果你希望优先级高的先出队那么高优对象的compareTo应该返回负数。初学者很容易把这个方向写反导致高优任务反而最后执行。排查时可以打印队列中的任务顺序先确认比较结果是否符合预期。另一个常见问题是消费者线程数量与核心线程数不等。上面配置里我启动消费者的方式是executor.execute(consumer)这会把核心线程全部占满。如果你手动提交了大量外部任务会发现外部任务没有空闲线程可以执行因为核心线程都在循环take()任务。这里其实有两种设计选择一是消费者即线程池的工作线程线程池不再对外接收其他任务二是把消费线程单独再建一个线程池而当前线程池只负责业务执行。实际项目中需要根据你的业务形态选择不能混用。6. 最佳实践与工程建议6.1 不要在大流量场景使用全量老化扫描示例中的老化方案在队列元素很少时没有问题但当队列积压达到几十万条时每次take()遍历全队列会消耗大量 CPU反而拖慢调度性能。工程上建议改用以下方案之一独立老化线程每隔固定时间扫描一次队列而不是每次取任务时扫描。分层队列按优先级拆成多个普通队列调度线程按照权重从高优到低优获取任务并记录低优队列等待时间。时间轮分桶按等待时间放入不同桶过期后统一升级。如果你的系统任务量超过单机处理能力优先考虑引入消息队列不要让单机队列承担过重的堆积职责。6.2 线程池参数需要根据任务特征调优线程数和队列容量的设置取决于你的任务属于 CPU 密集型还是 IO 密集型。CPU 密集型任务线程数建议接近 CPU 核数。IO 密集型任务线程数可以适当调大比如CPU核数 * 2或更高。队列容量必须是有界的。无界队列在流量突发时会导致内存暴涨进而引发 OOM。建议在你的压测环境里分别调整核心线程数和队列大小观察吞吐量、平均延迟、拒绝次数三个指标。不要仅仅照搬别人的配置。6.3 幂等设计是任务系统的底线本文使用的是本地内存Set只能解决单机内重复提交。生产环境的任务系统一定是多实例部署所以必须使用分布式幂等存储。推荐方案Redis使用SETNX命令任务 ID 作为 Key首次设置成功才允许执行。数据库唯一索引在任务表中给task_id加唯一约束插入冲突则说明已处理。消息队列自带去重比如 RocketMQ 的消息幂等消费需要业务方配合但框架不会自动去重。同时必须提醒一点调度框架的幂等只能防止“同一任务提交两次”不能消除业务逻辑本身的重复副作用。比如发送短信、扣减库存、转账等操作一定要在业务代码里再次做幂等校验。6.4 失败任务要区分“可重试”和“不可重试”并非所有异常都值得重试。网络超时、数据库连接短暂不可用、下游限流这些属于可重试参数错误、数据非法、业务状态不允许操作这些属于不可重试。如果对所有异常都重试不仅浪费资源还可能放大问题。建议在业务执行方法中封装一个异常分类public class BizException extends RuntimeException { private final boolean retryable; public BizException(String message, boolean retryable) { super(message); this.retryable retryable; } public boolean isRetryable() { return retryable; } }消费者在捕获异常时判断retryable字段只有可重试异常才进入重试逻辑。这样也可以避免无效重试把任务打到最大次数。6.5 增加可观测性任务调度系统属于异步链路一旦出问题很难直接感知。必须在代码中埋入足够的监控指标队列当前长度。每个优先级的任务数量。任务平均等待时间。任务执行成功率、失败率。每条任务的执行耗时分布。重试次数分布。在 Spring Boot 中可以通过Actuator暴露指标也可以使用Micrometer与 Prometheus 对接。如果没有现成的监控系统至少要在日志中打印关键指标方便线上排查。6.6 从单机队列到分布式消息队列的平滑演进当项目规模变大单机队列会逐渐成为瓶颈。建议演进路径如下阶段一单体应用任务量不大使用本文方案。阶段二任务量增长拆出独立的任务处理服务使用消息队列解耦生产者和消费者。阶段三需要严格全局优先级将高优任务单独发往一个高优先级的 Topic低优任务发往另一个 Topic消费者各自维护消费线程池并设置权重。在阶段三中不要试图在 Kafka 这类消息系统里实现“队列内动态插队”因为 Kafka 的分区顺序模型不支持这种操作。正确的做法是物理拆分 Topic用多个消费者组或消费优先级来解决。7. 总结与下一步回到刚开始那句话“苍生我要救钟离权我也要救。”在现实系统中这从来不是一个单选题。通过优先级队列 任务老化 重试 幂等这个组合方案我们可以让高优任务及时被处理同时低优任务也不会被永久饿死。本文通过一个完整的 Spring Boot 示例讲解了如何用PriorityBlockingQueue 自定义compareTo实现高优任务插队。如何用任务老化机制避免低优任务饿死。如何设计任务状态机让每个任务都有清晰的状态流转。如何在失败时用指数退避策略重试避免打垮下游。如何用任务 ID 做幂等防止重复执行。遇到问题时的排查顺序和常见坑点。如果这是你第一次接触任务调度下一步可以继续学习ThreadPoolExecutor的原理、ScheduledExecutorService的定时调度然后逐步尝试把任务调度扩展到 RocketMQ 或 RabbitMQ 的分布式场景。在实际项目中优先关注三件事线程池参数是否经过压测、幂等是否覆盖所有关键任务、监控告警是否在任务积压时第一时间通知你。把这三件事做好任务系统基本就能稳定运行。如果这篇文章对你有帮助建议收藏备用下次设计任务调度系统时可以直接对照本文的代码和排查表来落地。
