实现 submit(key, task)：同 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()，否则会等待只能在自己结束后才运行的任务。

注入的执行器必须异步安排工作，不能使用会在提交线程直接执行业务的策略；默认构造器满足这个前提。真实服务还要限制活动键数量、总线程数和每键队列容量，并考虑拒绝、截止时间与重启恢复。按题意暂不限制总并发，不代表线程和内存可以无限增长。

参考资料