带有背压的执行人员服务

Question

我对 executorService >：

• 有限的同时处理任务（线程），最好
• 在所有线程被占用时可配置，随后提交任务获得排队， 。
• 队列应应用背压：如果提交了太多任务并且队列已满，则提交应 block 。
• 阻止提交可以具有超时：如果在给定时间内未提交任务，则应引发例外。

原因：任务的生产者可能会使执行器服务或队列不堪重负，如果队列中的任务太多，则会增加内存需求。背压将在这种情况下阻止生产者，直到处理能够追赶为止。制作人（一些来自Kafka的阅读，JDBC的其他阅读）可能会有开放的交易，因此提交新任务（如果阻止）也应该能够超时。

作为执行者服务实现，例如 threadpoolexecutor ，请使用 blockingqueue 来加入任务，我希望队列实际上具有这样的功能，可以排队至定义的任务数，然后阻止，直到工作线程从队列中拉出任务为止。相反，事实证明，当队列已满时，提交时用 recondedexecutionException 拒绝其他任务。这当然不是我想要的。

我想到了 executorService 的包装器，该包装使用 smaphore 来控制同时接受多少个任务，这也使我可以在全部时阻止（使用超时）获得了信号量的许可：

public class Pipeline {

    private final ExecutorService executorService;
    private final Semaphore semaphore;

    private final int queueTimeout;
    private final TimeUnit queueTimeoutUnit;

    public Pipeline(int maxConcurrent, int queueTimeout, TimeUnit queueTimeoutUnit) {
        semaphore = new Semaphore(maxConcurrent);
        executorService = Executors.newCachedThreadPool();

        this.queueTimeout = queueTimeout;
        this.queueTimeoutUnit = queueTimeoutUnit;
    }

    public <T> Future<T> submit(Callable<T> task) {
        try {
            boolean acquired = semaphore.tryAcquire(queueTimeout, queueTimeoutUnit);
            if (!acquired) {
                throw new RuntimeException("Timeout accepting task");
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return executorService.submit(() -> {
            try {
                return task.call();
            } finally {
                semaphore.release();
            }
        });
    }

    public void shutdown() {
        executorService.shutdown();
    }
}

这实际上有效，但似乎是一种常见的用例，可以在Java API中涵盖。我是否缺少一些内置功能？

Answer 1

如果您创建 threadpoolexecutor ，并且提供了固定的长度 arrayblockingquequequequequequequequequequequequequequequequequeque 和您自己的自定义 reciendedexecutionhandler 怎么办？

笔记！我本人没有尝试过。这只是一个主意。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.TimeUnit;

class MyHandler implements RejectedExecutionHandler {
    final BlockingQueue<Runnable> task_queue;
    final long timeout;
    final TimeUnit unit;

    public MyHandler(
        BlockingQueue<Runnable> task_queue,
        long timeout,
        TimeUnit unit
    ) {
        this.task_queue = task_queue;
        this.timeout = timeout;
        this.unit = unit;
    }

    @override
    public void rejectedExecution(
        Runnable task,
        ThreadPoolExecutor pool
    ) {
        boolean timed_out = false;
        try {
            timed_out = ! task_queue.offer(task, timeout, unit);
        }
        catch (Exception ex) {
            ...use your shameful imagination here...
        }
        if (timed_out) {
            throw new RejectedExecutionException("queue is full");
        }
    }
}

然后，创建执行程序：

BlockingQueue<Runnable> task_queue = new ArrayBlockingQueue<>(...);
ExecutorService pool = new ThreadPoolExecutor(
    corePoolSize,  maximumPoolSize,
    keepAliveTime, keepAliveTimeUnit,
    task_queue,
    new MyHandler(task_queue, submitTimeout, submitTimeoutUnit)
);

有点hack-y，因为它会弄乱执行人背后的队列，但是我的期望是，如果代码调用 pool.submit（task）> ，然后将调用处理程序的 recondeDexecution（）方法。它将等待指定的时间将任务添加到队列中。

如果成功，那么任务就会添加到队列中，每个人都很高兴。如果失败，则 decredExeCutionException 将为 pool.submit（task）的呼叫者传播，以捕获和处理，但是您认为合适。

Answer 2

这是标准库中缺少的内容，以及为什么LMAX iSPLOR库存在：

我的经验是，它使并发消息队列变得轻而易举 - 根本不需要线程代码。警告是，最低的延迟等待策略需要大量的CPU资源。