实现 submit(key, task):同 key 串行、不同 key 并行,保证提交与执行队列线程安全,暂不限制总并发数。
未标记
考查重点:按键串行调度与入队退出竞态。为每个key维护队列和运行标记,提交与启动原子衔接;单消费者逐个执行同key任务,退出时防止漏掉同时入队的新任务
本题目录
参考答案
核心回答
为每个 key 维护一个先进先出的任务队列,每个队列同时只安排一个工作线程顺序消费;不同 key 的工作线程可以并行执行。提交、队列状态和工作线程的启动/退出需要原子衔接,用户的 Runnable 则在锁外执行。
最容易漏掉的竞态是:旧工作线程看见队列为空、准备退出时,另一个线程刚好提交新任务。如果“空队列检查与移除工作状态”和“入队与启动”不在同一同步边界里,新任务可能入队成功却再也没有消费者。
Java JUC 实现
题目暂不限制总并发数,这里使用 newCachedThreadPool() 为活动的 key 安排工作线程。同一 key 的并发提交,以进入同步块的顺序作为接收顺序;FutureTask 保存每个任务的结果、异常和取消状态。
import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
public final class KeyedExecutor implements AutoCloseable {
private final Object monitor = new Object();
private final Map<String, ArrayDeque<FutureTask<Void>>> queues = new HashMap<>();
private final ExecutorService workers;
private boolean closed;
public KeyedExecutor() {
this(Executors.newCachedThreadPool());
}
// 注入的线程池必须异步调度,不能在提交线程直接运行用户任务。
KeyedExecutor(ExecutorService workers) {
this.workers = Objects.requireNonNull(workers);
}
public Future<?> submit(String key, Runnable task) {
Objects.requireNonNull(key);
Objects.requireNonNull(task);
FutureTask<Void> future = new FutureTask<>(task, null);
synchronized (monitor) {
if (closed) throw new RejectedExecutionException("executor closed");
ArrayDeque<FutureTask<Void>> existing = queues.get(key);
if (existing != null) {
existing.addLast(future);
return future;
}
ArrayDeque<FutureTask<Void>> queue = new ArrayDeque<>();
queue.addLast(future);
queues.put(key, queue);
try {
workers.execute(() -> drain(key, queue));
} catch (RuntimeException | Error failure) {
queues.remove(key, queue);
throw failure;
}
}
return future;
}
private void drain(String key, ArrayDeque<FutureTask<Void>> queue) {
while (true) {
FutureTask<Void> task;
synchronized (monitor) {
task = queue.pollFirst();
if (task == null) {
queues.remove(key, queue);
return;
}
}
task.run(); // 业务异常由 FutureTask 保存,不终止后续任务。
Thread.interrupted(); // 不把本次任务取消留下的中断标记传给下一任务。
}
}
@Override
public void close() {
synchronized (monitor) {
closed = true;
workers.shutdown(); // 已接收的按键队列继续排空。
}
}
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
return workers.awaitTermination(timeout, unit);
}
}
为什么不会漏任务或并发执行同 key
映射中的队列同时代表“这个 key 已有工作线程被安排”。提交发现已有队列,只追加任务;创建新队列时,入队、加入映射和启动工作线程在同一同步块内完成。若线程池拒绝安排新工作线程,删除刚建立的映射并把异常交给调用方,避免留下无人消费的队列。
工作线程在同步块内取任务;只有看到队列为空时才删除映射并退出。如果新提交先进入同步块,旧线程下一轮会取到它;如果退出先发生,新提交会建立新队列和工作线程。业务执行、FutureTask 的完成通知都在同步块外,不会占住全局队列锁。
FutureTask.run() 会记录业务异常,因此一个任务失败不会终止同键后续任务;未开始的任务被取消后运行时会跳过。前一个任务取消留下的中断标记在本次执行结束后清理,避免误传给下一任务。正在运行的任务若忽略中断,也不会被强行终止,同键后续任务仍需等待它结束。
关闭与工程边界
close() 停止接受新提交,对线程池进行平滑关闭;已经接收的内部队列由原工作线程继续排空。调用方可以通过 awaitTermination() 等待结束。不要在同一个 key 的任务内同步等待后续同键任务的 Future.get(),否则会等待只能在自己结束后才运行的任务。
注入的执行器必须异步安排工作,不能使用会在提交线程直接执行业务的策略;默认构造器满足这个前提。真实服务还要限制活动键数量、总线程数和每键队列容量,并考虑拒绝、截止时间与重启恢复。按题意暂不限制总并发,不代表线程和内存可以无限增长。
参考资料
- 国内:JavaGuide:哈希表:采用按业务key定位队列的映射结构,区分查找状态与任务排序的职责。
- 国内:JavaGuide:线程池:采用线程池工作线程与待执行队列的分工,解释示例的缓存线程池只负责安排每键工作线程。
- 国内:JavaGuide:线程池最佳实践:采用总线程数、队列容量与优雅关闭的约束,说明题意暂不限并发不等于生产配置无上限。
- 国内:JavaGuide:并发常见问题(中):采用同一同步边界保护复合状态的方式,将入队、启动、空队列退出之间的竞态关在锁内。
- 国外:Oracle Java 21:Executor:参考串行执行器将提交与执行分离的设计,并核对执行器可能同步运行与拒绝任务的契约。
- 国外:Oracle Java 21:FutureTask:核对
Runnable封装、异常保存、结果等待与取消后run()不再执行业务的语义。

