java - 通过超时优雅地停止周期性消费者队列的优雅方式

Question

我有一个包含多个生产者 - 一个消费者的队列。消费者定期运行并完全排空队列(之后不留下任何消息)。优雅的算法应该运行消费者并超时等待，或者如果消费者已经在运行则等待。

目前我们有这样的东西:

void stop(boolean graceful) {
        if (graceful && !checkAndStopDirectly()) {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(shutdownWaitInterval, shutdownWaitIntervalUnit)) {
                    log.warn("...");
                }
            } catch (InterruptedException e) {
                log.error("...", e);
            }
        } else {
            executor.shutdownNow();
        }

private boolean checkAndStopDirectly() {
    ExecutorService shutdownExecutor = Executors.newSingleThreadExecutor();
    try {
        return shutdownExecutor.submit(new Callable<Boolean>(){
            @Override
            public Boolean call() throws Exception {
                if (isAlreadyRan.compareAndSet(false, true)) {
                    try {
                        runnableDrainTask.run();
                    } finally {
                        isAlreadyRan.set(false);
                    }
                    return true;
                } else {
                    return false;
                }
            }
        }).get(shutdownWaitInterval, shutdownWaitIntervalUnit);

有人看到更优雅的方法吗？例如我正在寻找一种不使用附加 AtomicBoolean (isAlreadyRan) 或以时间间隔作为对象字段等的双重等待逻辑的方法。顺便说一句，毒丸图案浮现在我的脑海中......

Answer 1

您在这里谈论的是应用程序的正常关闭应该执行以下操作吗？

1. 等待队列耗尽，或者
2. 如果排水时间过长则超时

我可能需要了解如何排空队列，但如果您能够以可中断的方式做到这一点，则可以在获取 Future 时超时并尝试无论如何，shutdownNow(如果未完全耗尽则中断)；您觉得这怎么样？

ExecutorService pool = Executors.newSingleThreadExecutor();

public void stop() {
    try {
        pool.submit(new DrainTask()).get(100, MILLISECONDS);
    } catch (TimeoutException e) {
        // nada, the timeout indicates the queue hasn't drained yet
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } catch (Exception e) {
        // nada, defer to finally
    } finally {
        pool.shutdownNow();
    }
}

private class DrainTask implements Callable<Void> {
    @Override
    public Void call() throws Exception {
        while (!Thread.currentThread().isInterrupted())
            ; // drain away
        return null;
    }
}

compareAndSet 部分是否用于防止多个并发调用 stop？我想我更喜欢通用锁或使用synchronized。但是，如果发生冲突，则会抛出 ExecutionException，并且重复调用 shutdownNow 就可以了。

它有点依赖于 DrainTask 能够停止正在执行的操作以响应中断调用(因为 shutdownNow 将尝试在任何当前正在运行的任务上调用中断)线程)。