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

作者：袖梨 2026-07-21

本文介绍一种高效、可扩展的并发调度方案：为相同属性（如“color”）的事件分配独立队列并由专用工作线程串行处理，同时通过共享线程池实现全局并发数限制，兼顾顺序性、隔离性与资源利用率。

本文介绍一种高效、可扩展的并发调度方案：为相同属性（如“color”）的事件分配独立队列并由专用工作线程串行处理，同时通过共享线程池实现全局并发数限制，兼顾顺序性、隔离性与资源利用率。

在高吞吐事件处理场景中，常需满足“同组串行、跨组并发、全局限流”三重要求——例如按 color 字段分组，绿色事件必须严格按接收顺序依次执行，黄色事件亦然；但绿与黄之间无需顺序约束；且所有颜色的执行线程总数不能超过系统设定上限（如 8 个并发线程）。

直接改造 ThreadPoolExecutor 的任务队列（如自定义 BlockingQueue）来实现“跳过同色正在运行任务”的动态出队逻辑，不仅破坏线程池设计契约，还极易引发竞态、死锁或饥饿问题（如某色任务持续积压导致其他色长期得不到调度）。因此，推荐采用“分组队列 + 共享工作者池”的解耦架构：

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

  • 每个 color 对应一个线程安全队列（如 ConcurrentLinkedQueue<Event> 或 LinkedBlockingQueue<Event>），保证该色内事件 FIFO；
  • 一个中央调度器线程（Distributor）持续从原始事件源（如 Kafka、消息队列或生产者队列）读取事件，根据 event.color() 将其路由至对应颜色队列；
  • 固定大小的共享线程池（如 Executors.newFixedThreadPool(N)）负责消费所有颜色队列：每个工作线程循环尝试从任意非空队列中取任务（优先 oldest 队列头），执行前加锁标记“该 color 正在运行”，执行后释放。

? 示例实现（Java）

// 1. 分组队列容器private final ConcurrentMap<String, Queue<Event>> colorQueues = new ConcurrentHashMap<>();private final ReentrantLock lock = new ReentrantLock();private final Set<String> runningColors = ConcurrentHashMap.newKeySet();// 2. 工作线程任务（提交至共享线程池）Runnable workerTask = () -> {    while (!Thread.currentThread().isInterrupted()) {        Event event = null;        String color = null;        // 轮询所有队列，找到首个可执行的 oldest 事件（避免饿死）        for (Queue<Event> 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、跨色完全并发、全局线程数可控，且代码清晰、易于监控与调试。相比侵入式修改线程池队列，它更符合面向对象与关注点分离原则，是生产环境推荐的稳健实践。

相关文章

精彩推荐