短频快task的java解决方案

程序浅谈后端 2024-10-17

短频快task的java解决方案

看到这个标题，大家一定很好奇，感觉这是一个重复造轮子的事情。java明明已经提供了WorkStealingPool，本身是带窃取能力的。这里就需要讲一下背景。这里主要来自WorkStealingPool的能力缺陷。

WorkStealingPool的能力缺陷java

 代码解读
复制代码
public static ExecutorService newWorkStealingPool() {
    return new ForkJoinPool
        (Runtime.getRuntime().availableProcessors(),
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

java的WorkStealingPool本质是forkjoinpool。他是利用了forkjoinpool的窃取任务的能力，来达到窃取的效果。forkjoinpool在实现复杂场景的fork，join的过程，也带来了一个问题，就是他无法做到中断程序。例如task1，拆出了task2,task3,task4。现在task2正在被执行，task3还在队列里，task4已经被窃取走了。现在对整体任务1发起中断。正确的中断是task2中断，task3取消，task4中断。最终对于task1而言是被中断了。这里就有了一个问题，task4被窃取走了，怎么知道那个线程需要被中断呢？这个问题forkjoinpool也没实现。在forkjointask的实现过程中也体现了这点。java

 代码解读
复制代码
public boolean cancel(boolean mayInterruptIfRunning) {
    return (setCompletion(CANCELLED) & DONE_MASK) == CANCELLED;
}

在取消任务的方法中mayInterruptIfRunning直接没有被使用。注释中直接写了这个参数不生效。arduino

 代码解读
复制代码
Params:
mayInterruptIfRunning – this value has no effect in the default implementation because interrupts are not used to control cancellation.

所以自带的WorkStealingPool是无法发起中断的。只能发起取消。

普通线程池为什么满足不了场景

java自带的线程数是一个生产消费模型，生产者就是我们提交的任务，消费者线程池的线程，用来执行任务。这个从模型上看还是一个公平模型，那个线程执行完了就从队列里获取任务，这样算力能力也不会被浪费。但是这个地方有个前提，就是任务执行的时间一定是得大于从队列获取的时间。对于短频快的任务，获取任务的损耗就会变得特别明显。线程池的队列必须是一个阻塞队列。这里就涉及到锁的竞争,如下的代码是线程池获取任务的地方workQueue就是阻塞队列，poll，take都会涉及到锁的竞争。java

 代码解读
复制代码
try {
    Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
        workQueue.take();
    if (r != null)
        return r;
    timedOut = true;
} catch (InterruptedException retry) {
    timedOut = false;
}

了解了场景之后，我们想这个场景的优化方式，其实之一就是批处理，反正任务执行快，把多个任务合并成一个，这样每个线程执行的时候就是按照一批来的。锁竞争的的损耗被降低了。
上面的批量的方案运行良好的情况每个任务的执行时间都差不多。但是现实情况是无法保证的，可能因为一些不合理的操作，导致某一个任务的执行时间变长，或者某一个批次的每个任务都比其他的多一些，最终导致了整体的执行时间变长。在整体的观测线程池的线程运行情况，就会发现某一个线程一直在执行，但是其他的线程还有空闲，造成了线程算力的浪费。

这里在看窃取的能力，似乎更好的满足了这里的诉求。