Java生态中支持冲突任务串行执行的线程池或工作流方案咨询

针对你提到的「同一乘客订单必须串行执行、无冲突任务可并行」的需求，Java生态里有几种实用的实现方式，不需要依赖过重的框架：

1. 基于锁的轻量实现（最简单直接） #

核心思路是给每个冲突维度（比如乘客ID）分配一把锁，任务执行前先获取对应锁，执行完成后释放。这样同一乘客的任务会因锁竞争自动串行，不同乘客的任务可以并行执行。

示例代码：

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class OrderTaskExecutor {
    private final ExecutorService pool = Executors.newFixedThreadPool(10);
    private final ConcurrentHashMap<String, Lock> passengerLocks = new ConcurrentHashMap<>();

    public void submitOrderTask(String passengerId, Runnable task) {
        pool.submit(() -> {
            // 懒加载创建锁，同一乘客共享同一把锁
            Lock lock = passengerLocks.computeIfAbsent(passengerId, k -> new ReentrantLock());
            lock.lock();
            try {
                task.run();
            } finally {
                lock.unlock();
                // 可选：如果乘客不再有任务，移除锁节省资源
                if (lock.tryLock()) {
                    try {
                        passengerLocks.remove(passengerId);
                    } finally {
                        lock.unlock();
                    }
                }
            }
        });
    }
}

这个方案的优点是代码简洁、无额外依赖，缺点是如果同一乘客的任务大量积压，会占用线程池中的线程等待锁（不过线程池本身可以控制并发数）。

2. 自定义ThreadPoolExecutor扩展（更精准的任务调度） #

如果需要更精细地控制任务调度（比如线程空闲时主动筛选无冲突任务执行），可以扩展ThreadPoolExecutor，自定义等待队列并跟踪正在运行的任务的冲突键：

• 重写beforeExecute和afterExecute方法，维护一个「正在运行的冲突键集合」；
• 自定义BlockingQueue，提供一个能筛选无冲突任务的方法；
• 重写getTask方法，让线程从队列中选取不与当前运行任务冲突的任务执行。

示例核心逻辑：

public class ConflictAwareThreadPool extends ThreadPoolExecutor {
    private final Set<String> runningConflictKeys = Collections.synchronizedSet(new HashSet<>());
    private final ConflictAwareQueue<Runnable> taskQueue;

    public ConflictAwareThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new SynchronousQueue<>());
        this.taskQueue = new ConflictAwareQueue<>();
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        // 假设任务是带冲突键的自定义类型
        if (r instanceof ConflictTask) {
            runningConflictKeys.add(((ConflictTask) r).getConflictKey());
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (r instanceof ConflictTask) {
            runningConflictKeys.remove(((ConflictTask) r).getConflictKey());
        }
    }

    @Override
    protected Runnable getTask() throws InterruptedException {
        // 从自定义队列中获取无冲突的任务
        Runnable task = taskQueue.pollAvailable(runningConflictKeys);
        if (task == null) {
            // 没有可用任务时，调用父类逻辑等待
            return super.getTask();
        }
        return task;
    }

    @Override
    public void execute(Runnable command) {
        if (!(command instanceof ConflictTask)) {
            throw new IllegalArgumentException("Task must be ConflictTask");
        }
        taskQueue.add((ConflictTask) command);
        super.execute(command);
    }
}

// 自定义带冲突键的任务
interface ConflictTask extends Runnable {
    String getConflictKey();
}

// 自定义队列
class ConflictAwareQueue<T extends ConflictTask> {
    private final Queue<T> queue = new LinkedList<>();

    public synchronized void add(T task) {
        queue.add(task);
        notifyAll();
    }

    public synchronized T pollAvailable(Set<String> runningKeys) throws InterruptedException {
        while (true) {
            // 筛选出不冲突的任务
            for (Iterator<T> it = queue.iterator(); it.hasNext(); ) {
                T task = it.next();
                if (!runningKeys.contains(task.getConflictKey())) {
                    it.remove();
                    return task;
                }
            }
            // 没有可用任务时等待
            wait();
        }
    }
}

这个方案能更高效地利用线程资源，线程空闲时会主动挑选可执行的无冲突任务，而不是等待队列头部的冲突任务。

3. 借助工作流引擎（复杂场景） #

如果你的任务调度逻辑更复杂（比如有依赖关系、优先级等），可以考虑使用Camunda、Activiti这类工作流引擎，它们支持任务的排他性调度（同一业务键的任务串行执行），不过这类框架通常比较重，适合复杂业务场景。

内容的提问来源于stack exchange，提问作者gavenkoa