本文介绍一种高效、可扩展的并发调度策略，通过为每种“颜色”维护独立任务队列并复用有限线程资源，确保同色事件严格串行、跨色事件并行执行，同时全局受线程池容量约束。
本文介绍一种高效、可扩展的并发调度策略，通过为每种“颜色”维护独立任务队列并复用有限线程资源，确保同色事件严格串行、跨色事件并行执行，同时全局受线程池容量约束。
在高吞吐事件处理场景中（如实时日志分析、订单状态更新、IoT设备指令分发），常需满足“同类别串行、跨类别并行、全局资源受限”的复合调度需求。直接改造 ThreadPoolExecutor 的工作队列 dequeue 逻辑（如自定义 BlockingQueue 实现“跳过冲突项”）不仅复杂、易出错，还可能破坏线程池的公平性与监控能力。更优解是采用分治+复用架构：以“颜色”为维度划分逻辑队列，再由少量共享工作线程按需消费——既保证语义正确性，又避免资源爆炸。
核心设计：颜色分桶 + 单线程消费者池
- ✅ 每个颜色独占一个 ConcurrentLinkedQueue
或 LinkedBlockingQueue ：天然支持 FIFO 插入与无锁 peek/poll，避免同色竞争。 - ✅ 统一调度器（Distributor）：单线程接收原始事件流，解析 color 字段，将任务路由至对应颜色队列。
- ✅ 固定大小的工作线程池（如 Executors.newFixedThreadPool(N)）：每个线程循环监听所有颜色队列（或使用轮询/优先级策略），仅当目标队列非空且当前无该颜色活跃任务时才取任务执行。
// 示例：轻量级调度器实现（简化版）
public class ColorAwareScheduler {
private final Map<string queue>> colorQueues = new ConcurrentHashMap();
private final ExecutorService workerPool;
private final Set<string> runningColors = ConcurrentHashMap.newKeySet();
public ColorAwareScheduler(int poolSize) {
this.workerPool = Executors.newFixedThreadPool(poolSize);
// 启动 N 个消费者线程
for (int i = 0; i new ConcurrentLinkedQueue()).add(task);
}
private void consumeNextTask() {
while (!Thread.currentThread().isInterrupted()) {
Runnable task = findOldestEligibleTask();
if (task != null) {
try {
task.run();
} finally {
// 执行完毕后释放颜色锁
String color = getColorFromTask(task); // 需业务提供提取逻辑
runningColors.remove(color);
}
} else {
LockSupport.parkNanos(1_000_000L); // 短暂休眠避免忙等
}
}
}
private Runnable findOldestEligibleTask() {
// 按插入顺序遍历所有队列（可优化为优先队列维护队首时间戳）
for (Queue<runnable> queue : colorQueues.values()) {
if (!queue.isEmpty()) {
Runnable candidate = queue.peek();
String color = getColorFromTask(candidate);
if (runningColors.add(color)) { // CAS 成功即抢占该颜色
return queue.poll(); // 安全移除
}
}
}
return null;
}
private String getColorFromTask(Runnable task) {
// 实际中可通过封装类（如 ColorTask）或反射获取；推荐前者
if (task instanceof ColorTask) {
return ((ColorTask) task).getColor();
}
throw new IllegalArgumentException("Task must implement ColorTask");
}
}</runnable></string></string>
关键注意事项
- 避免饥饿：若某颜色持续高频提交，可能长期占用线程。建议在 findOldestEligibleTask() 中引入时间戳优先级（如记录各队列首任务入队时间），优先调度等待最久的任务。
- 内存安全：ConcurrentLinkedQueue 无界，需配合背压机制（如 Semaphore 限流）或监控告警，防止 OOM。
- 优雅关闭：需确保所有队列为空、所有运行中任务完成后再 shutdown workerPool，推荐使用 CountDownLatch 或 CompletableFuture.allOf() 协调。
- 扩展性增强：当颜色数量极大（如百万级用户ID作为color），可用 ConcurrentHashMap 分段哈希 + 动态队列回收（空闲超时自动清理），避免内存泄漏。
该方案摒弃了侵入线程池底层的高风险做法，以清晰的职责分离（分发 vs 执行）、可控的资源边界（固定线程数）和自然的串行保障（单队列单消费者语义），成为生产环境首选。
