多ThreadPoolExecutor类型共用全局任务队列实现呼叫中心优先级接单问询

问题背景 #

我正在完成一项书本中的OOP编程练习，设计呼叫中心代码。呼叫中心有Respondent（接线员）、Manager（经理）、Director（总监）三类应答者，存在接单优先级顺序。目前仅能实现单一类型对象处理队列中的呼叫任务，需求是让三类应答者对应的ThreadPoolExecutor都能轮询同一个全局任务队列获取任务，尝试过创建多个ThreadPoolExecutor实例共用同一队列，现寻求可行实现建议。

现有代码片段 #

Centre类main方法

public class Centre {

    public static void main(String[] args) {

        // initialize resources
        CallScheduler scheduler = new CallScheduler();
        scheduler.initCallQueue();

        // set up roster here
        int n_respondents = 6;
        int n_managers = 5;
        int n_directors = 3;

        scheduler.initRosterExecutors(n_respondents, n_managers, n_directors);

        // listen for incoming client requests
        try {

            ServerSocket server = new ServerSocket(SERVICE_PORT);
            System.out.println("Call Centre started.");

            for (;;) {

                Socket nxt_cal = server.accept();

                System.out.println("Received call from " + nxt_cal.getInetAddress() + ":"
                        + nxt_cal.getPort());

                scheduler.dispatchCall(nxt_cal);
                Logger.global.fine("Dispatched Call.");

                // Close the connection
                nxt_cal.close();
            }
        }
        // 此处省略异常捕获代码
    }
}

initRosterExecutors方法

public static void initRosterExecutors(int res, int man, int dir) {

    final int first_to_the_call = 1;

    res_executor =
            new RespondentExecutor(first_to_the_call, res, 5, TimeUnit.SECONDS,
                    bqu, new RespondentExecutor.AbortPolicy());

    res_executor.prestartAllCoreThreads();

    man_executor =
            new RespondentExecutor(first_to_the_call, man, 5, TimeUnit.SECONDS,
                    bqu, new RespondentExecutor.AbortPolicy());

    man_executor.prestartAllCoreThreads();

    dir_executor =
            new RespondentExecutor(first_to_the_call, dir, 5, TimeUnit.SECONDS,
                    bqu, new RespondentExecutor.AbortPolicy());

    dir_executor.prestartAllCoreThreads();

}

核心问题分析 #

直接让多个ThreadPoolExecutor共享同一无界队列，会导致所有线程无差别抢任务，完全无法体现"接单优先级"的业务逻辑——比如总监线程可能会抢走本该由接线员处理的任务，违背呼叫中心的分级服务规则。

可行实现方案 #

方案1：优先级队列+任务权限过滤

• 给呼叫任务添加级别标记（普通、升级、紧急），用PriorityBlockingQueue作为全局任务队列，按任务级别排序。
• 自定义不同类型的ThreadPoolExecutor，重写getTask()方法，让线程只处理对应级别及以下的任务：
  • 接线员线程：仅处理普通级别任务，若队列中只有高级任务则等待
  • 经理线程：可处理普通、升级任务
  • 总监线程：处理所有级别任务
• 示例代码（重写getTask）：

@Override
protected Runnable getTask() {
    while (!isShutdown()) {
        Runnable task = super.getTask();
        if (task == null) return null;
        
        CallTask callTask = (CallTask) task;
        // 接线员线程只处理RESPONDENT级任务
        if (callTask.getLevel() == CallLevel.RESPONDENT) {
            return task;
        }
        // 非权限内任务放回队列，短暂等待后重试
        getQueue().offer(task);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }
    return null;
}

方案2：分层队列+溢出调度（更贴合业务场景）

这是呼叫中心的经典实现模式，完全避免共享队列的抢任务问题：

• 为三类应答者分别创建独立的任务队列：respondentQueue、managerQueue、directorQueue
• 调度逻辑：
  1. 新呼叫先放入respondentQueue，由接线员线程池处理
  2. 若respondentQueue已满且接线员线程全部忙碌，将任务转入managerQueue
  3. 若managerQueue也满且经理线程忙碌，再转入directorQueue
• 每个线程池仅消费自己对应的队列，保证低级任务优先由低级别人员处理。

方案3：自定义任务调度器

单独实现一个调度器类，持有全局任务队列和三类线程池，调度器负责按优先级将任务分配给对应线程池：

• 调度器实时监控各线程池的忙碌状态
• 新任务进来时，优先分配给空闲的接线员；若接线员全忙，再分配给经理，以此类推
• 这种方式灵活性最高，可随时调整优先级规则。

代码调整示例（方案2） #

1. 定义任务级别和包装类：

public enum CallLevel {
    RESPONDENT, MANAGER, DIRECTOR
}

public class CallTask implements Runnable {
    private Socket callSocket;
    private CallLevel level;

    public CallTask(Socket callSocket, CallLevel level) {
        this.callSocket = callSocket;
        this.level = level;
    }

    // getter、setter方法
    @Override
    public void run() {
        // 具体呼叫处理逻辑
    }
}

2. 修改调度器的队列初始化和任务分发：

// 调度器类中维护三个队列
private BlockingQueue<Runnable> respondentQueue = new ArrayBlockingQueue<>(10);
private BlockingQueue<Runnable> managerQueue = new ArrayBlockingQueue<>(5);
private BlockingQueue<Runnable> directorQueue = new ArrayBlockingQueue<>(3);

public void initRosterExecutors(int res, int man, int dir) {
    res_executor = new ThreadPoolExecutor(1, res, 5, TimeUnit.SECONDS, respondentQueue);
    res_executor.prestartAllCoreThreads();

    man_executor = new ThreadPoolExecutor(1, man, 5, TimeUnit.SECONDS, managerQueue);
    man_executor.prestartAllCoreThreads();

    dir_executor = new ThreadPoolExecutor(1, dir, 5, TimeUnit.SECONDS, directorQueue);
    dir_executor.prestartAllCoreThreads();
}

public void dispatchCall(Socket socket) {
    // 尝试放入接线员队列，失败则转经理队列，再失败转总监队列
    if (!respondentQueue.offer(new CallTask(socket, CallLevel.RESPONDENT), 1, TimeUnit.SECONDS)) {
        if (!managerQueue.offer(new CallTask(socket, CallLevel.MANAGER), 1, TimeUnit.SECONDS)) {
            directorQueue.offer(new CallTask(socket, CallLevel.DIRECTOR));
        }
    }
}

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