按属性分组的线程安全串行执行与全局并发控制实现

在实际生产环境中，我们经常面临一个并发调度难题：如何确保按“color”字段分组后，同一组事件严格按序执行，不同组事件能并行处理，并且整个系统的并发线程数受到精准控制？这实际上是一个平衡顺序性、隔离性和资源利用率的经典问题。

在高吞吐量事件处理场景下，通常需要同时实现“同组串行、跨组并发、全局限流”三大目标。举例来说，假设事件按照color字段分组：绿色事件必须严格按接收顺序依次执行，黄色事件同样如此；但绿色和黄色之间无执行顺序约束，可以并行处理。同时，整个系统所有颜色的执行线程总数必须控制在预设上限内，例如8个线程。

直接修改ThreadPoolExecutor的任务队列，例如自定义BlockingQueue，尝试实现“跳过同色正在运行的任务”的动态出队逻辑，这不但会破坏线程池原有的设计契约，还极易引发竞态条件、死锁或饥饿问题——例如某一颜色事件持续积压，其他颜色长期得不到调度。因此，业界更推荐采用解耦架构：分组队列 + 共享工作者池。

核心设计：分组队列与统一调度器

• 每个color配备一个线程安全队列（如ConcurrentLinkedQueue或LinkedBlockingQueue），确保该颜色内部事件严格遵循FIFO（先进先出）顺序；
• 一个中央调度器线程（Distributor）不断从原始事件源（例如Kafka、消息队列或生产者队列）读取事件，并根据event.color()将其路由到对应的颜色队列中；
• 一个固定大小的共享线程池（例如Executors.newFixedThreadPool(N)）负责消费所有颜色队列。每个工作线程循环尝试从任意非空队列中取任务（优先取队列头部较旧的事件），执行前锁定标记“该color正在运行”，执行完成后立即释放锁。

示例实现（Java）

// 1. 分组队列容器
private final ConcurrentMap> colorQueues = new ConcurrentHashMap<>();
private final ReentrantLock lock = new ReentrantLock();
private final Set runningColors = ConcurrentHashMap.newKeySet();

// 2. 工作线程任务（提交至共享线程池）
Runnable workerTask = () -> {
    while (!Thread.currentThread().isInterrupted()) {
        Event event = null;
        String color = null;
        // 轮询所有队列，找到首个可执行的 oldest 事件（避免饿死）
        for (Queue queue : colorQueues.values()) {
            if (!queue.isEmpty()) {
                event = queue.peek(); // 先看一眼，不移除
                if (event != null && !runningColors.contains(event.color())) {
                    color = event.color();
                    event = queue.poll(); // 确认后出队
                    break;
                }
            }
        }
        if (event == null) {
            Thread.sleep(10); // 短暂让出 CPU
            continue;
        }
        // 标记 color 正在运行
        runningColors.add(color);
        try {
            event.execute(); // 执行业务逻辑
        } finally {
            runningColors.remove(color); // 必须确保释放
        }
    }
};

// 启动 N 个 worker 线程
ExecutorService workers = Executors.newFixedThreadPool(8);
for (int i = 0; i < 8; i++) {
    workers.submit(workerTask);
}

关键注意事项与优化建议

• 避免锁竞争：runningColors采用ConcurrentHashMap.newKeySet()代替synchronized块，可显著提升并发读写性能；
• 防止任务丢失：peek() + poll()组合需确保原子性。若poll()返回null（被其他线程抢先），则需重试；
• 公平性保障：轮询所有队列（而非固定顺序）可缓解某些颜色长期积压的问题。进阶方案可引入优先级队列，按队列头部时间戳排序；
• 资源清理：空队列可定期清理，例如colorQueues.entrySet().removeIf(e -> e.getValue().isEmpty() && !runningColors.contains(e.getKey()))，防止内存泄漏；
• 扩展性：支持动态增加或销毁color，无需重启服务即可生效。

这套方案天然满足所有原始需求：同色严格FIFO、跨色完全并发、全局线程数可控，且代码结构清晰，便于监控与调试。相比侵入式修改线程池队列，它更符合面向对象和关注点分离原则，是生产环境中值得推荐的稳健实践。