设计一个支持任务优先级且防止饥饿的线程池

题目

设计一个支持任务优先级且防止饥饿的线程池

信息

• 类型：问答
• 难度：⭐⭐

考点

线程池配置,任务优先级调度,线程饥饿预防,并发控制

快速回答

要实现支持任务优先级且防止饥饿的线程池，需要：

• 使用PriorityBlockingQueue作为工作队列
• 自定义Comparator实现任务优先级排序
• 实现公平调度策略防止低优先级任务饥饿
• 合理配置线程池参数（核心/最大线程数等）
• 使用ThreadPoolExecutor的钩子方法进行监控

原理说明

标准线程池使用FIFO队列，无法处理优先级任务。通过PriorityBlockingQueue和自定义比较器可实现任务优先级调度，但需解决低优先级任务可能被无限延迟的问题（饥饿）。解决方案是结合时间戳实现老化机制（Aging），随着等待时间增加而提升任务优先级。

代码实现

1. 定义优先级任务

class PriorityTask implements Runnable {
    private final Runnable task;
    private final int priority; // 1-10, 10最高
    private final long createTime;

    public PriorityTask(Runnable task, int priority) {
        this.task = task;
        this.priority = priority;
        this.createTime = System.currentTimeMillis();
    }

    // 计算动态优先级（等待越久优先级越高）
    public int getCurrentPriority() {
        long waitTime = System.currentTimeMillis() - createTime;
        if (waitTime > 5000) return 10; // 等待超5秒升为最高
        if (waitTime > 2000) return Math.min(10, priority + 2);
        return priority;
    }

    @Override
    public void run() {
        task.run();
    }
}

2. 创建线程池

// 自定义比较器（考虑动态优先级）
Comparator<Runnable> comparator = (r1, r2) -> {
    PriorityTask t1 = (PriorityTask) r1;
    PriorityTask t2 = (PriorityTask) r2;
    return Integer.compare(t2.getCurrentPriority(), t1.getCurrentPriority());
};

// 创建线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4, // 核心线程数
    8, // 最大线程数
    60, TimeUnit.SECONDS,
    new PriorityBlockingQueue<>(20, comparator), // 优先级队列
    new CustomThreadFactory(), // 自定义线程工厂
    new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
);

// 提交不同优先级任务
executor.execute(new PriorityTask(lowPriorityTask, 3));
executor.execute(new PriorityTask(highPriorityTask, 8));

最佳实践

• 防饥饿策略：通过getCurrentPriority()实现动态优先级提升
• 队列监控：重写beforeExecute/afterExecute记录任务状态
• 线程工厂：给线程命名便于监控（如pool-priority-thread-%d）
• 拒绝策略：高优先级任务被拒绝时可记录日志或暂存重试

常见错误

• 优先级反转：高优先级任务依赖低优先级资源导致阻塞 → 使用锁优先级继承
• 队列无限增长：未设置合理队列容量导致OOM → 根据系统负载设置上限
• 线程泄漏：未正确处理异常导致工作线程退出 → 使用afterExecute捕获异常
• 静态优先级：未实现老化机制导致低优先级任务饥饿

扩展知识

• 监控指标：通过getQueue().size()监控队列堆积，getActiveCount()监控活动线程
• 动态调参：通过setCorePoolSize()动态调整核心线程数应对流量波动
• 替代方案：考虑ForkJoinPool的work-stealing机制处理CPU密集型任务
• 框架集成：Spring的ThreadPoolTaskExecutor可简化配置管理