如何通过Map跟踪多队列间任务的流转状态？

这问题我之前在做任务编排系统时碰到过，手动维护状态不仅冗余还容易出一致性问题，尤其是后续要扩展不同类型队列的时候，简直是噩梦。分享几个可落地且能无缝扩展的思路：

1. 队列装饰器（最推荐，贴合你的需求） #

核心思路是把状态追踪逻辑和队列本身绑定，写一个通用的队列装饰器类，实现Queue接口，内部包装真实队列（不管是ArrayBlockingQueue还是DelayQueue都能套），在队列的add、take等关键操作时自动更新状态。这样业务代码完全不用关心状态维护，只和装饰后的队列交互就行。

举个Java的实现例子（其他语言思路一致）：

// 先定义状态枚举
enum TaskStatus { IN_FIRST, AFTER_FIRST, IN_SECOND, AFTER_SECOND, ABSENT }

// 通用状态追踪队列装饰器
public class StateTrackingQueue<T> implements Queue<T> {
    private final Queue<T> delegateQueue; // 真实队列
    private final ConcurrentMap<T, TaskStatus> statusMap; // 线程安全的状态映射
    private final TaskStatus statusOnAdd; // 加入队列时的状态
    private final TaskStatus statusOnRemove; // 移出队列时的状态

    public StateTrackingQueue(Queue<T> delegate, ConcurrentMap<T, TaskStatus> statusMap,
                              TaskStatus addStatus, TaskStatus removeStatus) {
        this.delegateQueue = delegate;
        this.statusMap = statusMap;
        this.statusOnAdd = addStatus;
        this.statusOnRemove = removeStatus;
    }

    @Override
    public boolean add(T item) {
        boolean success = delegateQueue.add(item);
        if (success) {
            statusMap.put(item, statusOnAdd);
        }
        return success;
    }

    @Override
    public T take() throws InterruptedException {
        T item = delegateQueue.take();
        statusMap.put(item, statusOnRemove);
        return item;
    }

    // 其他Queue方法（比如offer、poll等）都委托给delegateQueue，按需处理状态
    @Override
    public boolean offer(T t) {
        boolean success = delegateQueue.offer(t);
        if (success) {
            statusMap.put(t, statusOnAdd);
        }
        return success;
    }
}

然后你的业务代码就简化成这样，完全看不到状态维护的冗余代码：

// 初始化真实队列
Queue<Item> firstQueue = new ArrayBlockingQueue<>(100);
Queue<Item> secondQueue = new DelayQueue<>(); // 后续换其他队列直接改这里就行

// 包装成带状态追踪的队列
ConcurrentMap<Item, TaskStatus> statusMap = new ConcurrentHashMap<>();
Queue<Item> trackedFirst = new StateTrackingQueue<>(firstQueue, statusMap, TaskStatus.IN_FIRST, TaskStatus.AFTER_FIRST);
Queue<Item> trackedSecond = new StateTrackingQueue<>(secondQueue, statusMap, TaskStatus.IN_SECOND, TaskStatus.AFTER_SECOND);

// 处理线程的代码变得异常简洁
while (true) { 
    Item i = trackedFirst.take(); 
    process(i); // 你的业务处理逻辑
    trackedSecond.add(i); 
}

为什么这个方案适合你？ #

• 消除代码冗余：状态更新逻辑被封装在装饰器里，业务代码只关注任务流转
• 解决状态不一致：队列操作和状态更新在同一个方法内完成，用ConcurrentMap或者给装饰器方法加锁（如果需要强一致），完全消除时间窗口
• 无缝扩展队列：后续加PriorityQueue、LinkedBlockingQueue等，只需要用装饰器包装就行，不用改任何业务逻辑

2. 让任务对象自身管理状态 #

如果你的Item类是可控的，可以把状态直接存在任务对象里，配合队列装饰器调用任务的状态更新方法。这种方式更直观，不用单独维护statusMap：

public class Item {
    private volatile TaskStatus status = TaskStatus.ABSENT; // volatile保证线程可见性

    public void enterFirstQueue() {
        this.status = TaskStatus.IN_FIRST;
    }

    public void leaveFirstQueue() {
        this.status = TaskStatus.AFTER_FIRST;
    }

    // 对应其他队列的enter/leave方法
    public TaskStatus getStatus() {
        return status;
    }
}

然后修改装饰器，在add和take时调用任务的对应方法就行，比如：

@Override
public boolean add(Item item) {
    boolean success = delegateQueue.add(item);
    if (success) {
        item.enterFirstQueue();
    }
    return success;
}

这个方案的好处是状态和任务强绑定，避免了statusMap可能出现的内存泄漏（比如任务完成后没移除映射），但需要修改任务类的结构。

3. 事件驱动的状态追踪（适合复杂流转场景） #

如果后续任务流转逻辑会变得更复杂（比如多队列跳转、分支流转），可以用事件驱动的方式：队列操作触发状态事件，一个专门的状态处理器订阅事件并更新状态。

比如用简单的事件总线实现：

// 定义状态变更事件
public class TaskStateEvent {
    private final Item item;
    private final TaskStatus newStatus;

    public TaskStateEvent(Item item, TaskStatus newStatus) {
        this.item = item;
        this.newStatus = newStatus;
    }

    // getters
}

// 事件总线（可以用Guava EventBus或者自己实现简单版）
public class EventBus {
    private final List<EventListener> listeners = new CopyOnWriteArrayList<>();

    public void register(EventListener listener) {
        listeners.add(listener);
    }

    public void post(TaskStateEvent event) {
        for (EventListener listener : listeners) {
            listener.handle(event);
        }
    }
}

// 状态更新处理器
public class StateUpdateListener implements EventListener {
    private final ConcurrentMap<Item, TaskStatus> statusMap;

    public StateUpdateListener(ConcurrentMap<Item, TaskStatus> statusMap) {
        this.statusMap = statusMap;
    }

    @Override
    public void handle(TaskStateEvent event) {
        statusMap.put(event.getItem(), event.getNewStatus());
    }
}

然后在队列装饰器里发布事件：

@Override
public T take() throws InterruptedException {
    T item = delegateQueue.take();
    eventBus.post(new TaskStateEvent(item, TaskStatus.AFTER_FIRST));
    return item;
}

这种方式解耦性最强，后续加新的队列或状态，只需要加对应的事件和处理器，完全不影响业务代码，但复杂度稍高，适合复杂场景。

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