本文介绍一种高效、可扩展的线程调度策略——通过为每类属性（如“颜色”）维护独立任务队列，并由共享线程池中的工作者线程按需消费，确保同属性任务严格串行、跨属性任务并发执行，同时全局受线程数限制。
本文介绍一种高效、可扩展的线程调度策略——通过为每类属性（如“颜色”）维护独立任务队列，并由共享线程池中的工作者线程按需消费，确保同属性任务严格串行、跨属性任务并发执行，同时全局受线程数限制。
在高吞吐事件处理场景中，若要求“相同属性（如 color）的任务必须严格 FIFO 串行执行，而不同属性之间可并发”，直接改造 ThreadPoolExecutor 的队列出队逻辑（如自定义 BlockingQueue 的 poll()）不仅复杂、易出错，还可能破坏线程池内部状态（如 workers 计数、中断信号传递等），不推荐。
✅ 推荐方案：分发器 + 属性专属队列 + 共享工作线程池
核心思想是解耦“调度逻辑”与“执行逻辑”：
-
一个轻量级分发器线程（或主线程）负责接收原始事件流，按 color 哈希/映射到对应的 ConcurrentLinkedQueue
或 LinkedBlockingQueue ； -
每个颜色对应一个独立队列（可复用 ConcurrentHashMap
> 管理），保证同色任务天然有序； - 固定大小的 ThreadPoolExecutor（例如 corePoolSize = maxConcurrency）不直接消费原始队列，而是从所有颜色队列中轮询或优先选择最老待处理任务（见下文优化）；
- 关键约束：每个颜色队列最多允许一个任务处于“运行中”状态 —— 这通过为每个颜色维护一个原子标记（如 AtomicBoolean processing）实现，仅当 !processing.compareAndSet(false, true) 成功时才出队并提交执行；执行完毕后重置标记。
以下是精简可运行示例：
public class ColorAwareExecutor {
private final ExecutorService workerPool;
private final ConcurrentHashMap<string queue>> colorQueues = new ConcurrentHashMap();
private final ConcurrentHashMap<string atomicboolean> colorLocks = new ConcurrentHashMap();
public ColorAwareExecutor(int maxConcurrency) {
this.workerPool = Executors.newFixedThreadPool(maxConcurrency);
}
public void submit(Event event) {
String color = event.getColor();
colorQueues.computeIfAbsent(color, k -> new ConcurrentLinkedQueue()).add(() -> event.handle());
// 触发或确保有线程尝试消费该 color 队列
tryScheduleForColor(color);
}
private void tryScheduleForColor(String color) {
AtomicBoolean lock = colorLocks.computeIfAbsent(color, k -> new AtomicBoolean());
if (lock.compareAndSet(false, true)) {
Queue<runnable> queue = colorQueues.get(color);
if (queue != null && !queue.isEmpty()) {
Runnable task = queue.poll();
if (task != null) {
workerPool.submit(() -> {
try {
task.run();
} finally {
// 任务完成，释放锁，触发下一轮调度
lock.set(false);
tryScheduleForColor(color); // 继续消费同色队列
}
});
} else {
lock.set(false);
}
} else {
lock.set(false);
}
}
}
public static class Event {
private final String color;
private final Runnable handler;
public Event(String color, Runnable handler) {
this.color = color;
this.handler = handler;
}
public String getColor() { return color; }
public void handle() { handler.run(); }
}
}</runnable></string></string>
? 关键优势与注意事项：
- ✅ 强顺序性：同一 color 下任务绝对 FIFO，无竞态；
- ✅ 资源可控：总并发数 = ThreadPoolExecutor 的 corePoolSize，不受颜色数量影响；
- ✅ 低延迟响应：新事件到达即入队，无需等待全局队列扫描；
- ⚠️ 避免过度创建队列：对稀疏颜色（如百万种 ID），可使用 String.hashCode() % N 分桶（N ≈ 64~256），在桶内做串行，平衡隔离性与内存开销；
- ⚠️ 异常处理：务必在 finally 中重置 AtomicBoolean，否则该颜色将永久阻塞；
- ? 进阶优化：如需严格全局“最老事件优先”，可在分发器中维护一个 PriorityQueue
（按时间戳排序），但仅用于触发调度，实际执行仍走颜色队列，兼顾正确性与性能。
该模式已被 Kafka Consumer、Netty EventLoop Group 等成熟系统采用，兼顾简洁性、健壮性与可维护性，是解决“分组有序+全局限流”问题的工业级标准实践。
