java - 异构任务集的动态优先级

Question

我有一堆重复的任务要安排。他们查询数据库以找出要做什么，然后执行一些操作，如统计更新、发送电子邮件、获取文件并导入它们。目前，可能有十个，而且这个数字预计会增长很多。我没有任何时间限制，实际上，我的工作是选择一种算法，这样就不会有人提示了。 :D

目前，我正在使用线程和定期计划任务的临时组合，例如

• 对于最重要的任务，有一个自己的线程在空闲时回落到短暂的 sleep (当新的重要工作到达时，它可以从中唤醒)。
• 另一项重要任务在其自己的线程中每小时安排一次
• 中等重要性的任务会定期安排以“填补漏洞”，因此在任何时刻可能只有其中一个任务在运行
• 最不重要的任务都由一个专用线程处理

它目前似乎运行良好，但它不是面向 future 的，并且由于以下原因感觉不合适:

• 由于最不重要任务的队列可能会增长很多，因此此类任务可能会无限期延迟。
• 填坑可能会出错，同时可能有很多任务在运行。
• 在任何给定时刻运行的任务数量应取决于服务器负载。 (*)

(*) 它主要是一个网络服务器，服务请求实际上是最高优先级的。获得单独的服务器无济于事，因为瓶颈通常是数据库。目前，它运行良好，但我正在寻找更好的解决方案，因为我们希望负载在一两年内增长 100 倍。

我的想法是在工作延迟太多时提高工作的优先级。比如有统计每小时运行一次，延迟几个小时没什么大不了的，但不应该是一整天，也不能是一整周。

我很乐意将我所有的 AbstractExecutionThreadService 和 AbstractScheduledService 替换为如下工作方式:

• 无论如何，立即启动优先级最高的任务。
• 仅当总负载“小”时才启动中优先级任务。
• 仅当系统“大部分空闲”时才启动优先级最低的任务。
• 使用提供的公式提高延迟任务的优先级。

这听起来确实很模糊，让它更精确是我要问的一部分。我的竞争目标是

• 切勿不必要地拖延重要任务。
• 永远不要让过多并发运行的任务拖慢服务器的速度。

没有硬性截止日期，也没有必要尽量减少使用的线程数。我不坚持要求解决方案完全按照我的描述进行，我不是在寻找图书馆(我也不是坚持要重新发明轮子)。我认为类似 cron 的调度程序不是正确的解决方案。

Answer 1

使用 ExecutorService模型，重新排序执行程序任务的经典解决方案是创建一个 ThreadPoolExecutor用PriorityBlockingQueue向它提供任务 - 如上所述 here .

然而，需要安排任务以及对其进行调整。 ScheduledThreadPoolExecutor使用内部自定义 BlockingQueue在计划准备就绪时提供任务，但我认为您很清楚，进一步定制并不容易。

一目了然，DelayQueue看起来它完全符合要求——它可以优先考虑下一个 Delayed要素或任务。这处理了 Delayed.getDelay() 的延迟决定关于它是否准备就绪。

这个计划的美中不足是当你试图传递类似 DelayQueue<DelayedRunnable> 的东西时进入 ThreadPoolExecutor 的构造函数.这将只接受 BlockingQueue<Runnable> , 不是 BlockingQueue<? extends Runnable> .

解决此问题的一种方法是创建 BlockingQueue<Runnable> 的最小实现委托(delegate)给 BlockingQueue .基础知识在这里:

public class BlockingDelayQueue extends AbstractQueue<Runnable>
        implements BlockingQueue<Runnable> {

    private final DelayQueue<DelayedRunnable> delayQueue;

    public BlockingDelayQueue(DelayQueue<DelayedRunnable> delayQueue) {
        this.delayQueue = delayQueue;
    }

    @Override
    public boolean isEmpty() {
        return delayQueue.isEmpty();
    }

    @Override
    public Runnable poll(long timeout, TimeUnit unit)
            throws InterruptedException {
        DelayedRunnable delayedRunnable = delayQueue.poll(timeout, unit);

        if (delayedRunnable == null)
            return null;
        return delayedRunnable.getCommand();
    }
    ...
}

实验版DelayedRunnable用来证明那里的想法使用一个简单的Priority检查执行者“忙碌”的枚举:

LOW {
    boolean isReady(ThreadPoolExecutor executor) {
        return executor.getActiveCount() == 0;
    }
},
MEDIUM {
    boolean isReady(ThreadPoolExecutor executor) {
        return executor.getActiveCount() <= 1;
    }
},
HIGH {
    boolean isReady(ThreadPoolExecutor executor) {
        return true;
    }
};

哪个DelayedRunnable.getDelay()然后可以检查:

@Override
public long getDelay(TimeUnit unit) {
    long millis;
    if (!priority.isReady(executor))
        millis = 1000;
    else
        millis = time - System.currentTimeMillis();
    return unit.convert(millis, TimeUnit.MILLISECONDS);
}

- 只要它不返回 <= 0如果priority还没有准备好。

这似乎运作良好，例如在这里启动标准的 2s sleep 任务...

    DelayedScheduler scheduler = new DelayedScheduler();
    scheduler.schedule(task("Low 1"), 1, TimeUnit.SECONDS, Priority.LOW);
    scheduler.schedule(task("Low 2"), 2, TimeUnit.SECONDS, Priority.LOW);
    scheduler.schedule(task("Low 3"), 3, TimeUnit.SECONDS, Priority.LOW);
    scheduler.schedule(task("Medium 1"), 1, TimeUnit.SECONDS, Priority.MEDIUM);
    scheduler.schedule(task("Medium 2"), 2, TimeUnit.SECONDS, Priority.MEDIUM);
    scheduler.schedule(task("Medium 3"), 3, TimeUnit.SECONDS, Priority.MEDIUM);
    scheduler.schedule(task("High 1"), 1, TimeUnit.SECONDS, Priority.HIGH);
    scheduler.schedule(task("High 2"), 2, TimeUnit.SECONDS, Priority.HIGH);
    scheduler.schedule(task("High 3"), 3, TimeUnit.SECONDS, Priority.HIGH);

...产生了正确的结果:

High 1 started at 1087ms
Medium 1 started at 1087ms
High 2 started at 2087ms
  Medium 1 ended at 3087ms
  High 1 ended at 3087ms
High 3 started at 3087ms
  High 2 ended at 4088ms
Medium 2 started at 4088ms
  High 3 ended at 5088ms
Medium 3 started at 5088ms
  Medium 2 ended at 6088ms
  Medium 3 ended at 7089ms
Low 1 started at 7089ms
  Low 1 ended at 9089ms
Low 2 started at 9089ms
  Low 2 ended at 11089ms
Low 3 started at 11089ms
  Low 3 ended at 13089ms

- 当只有一个高优先级任务在运行时允许中优先级任务，当没有其他任务运行时允许低优先级任务。

( DelayedScheduler 和 GitHub 上的其他看不见的位)。