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

【问题标题】：elegant way of graceful stopping queue with periodic consumers by timeout通过超时优雅地停止队列与周期性消费者的优雅方式
【发布时间】：2011-08-12 04:30:36
【问题描述】：

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

目前我们有这样的：

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) 或具有时间间隔的双重等待逻辑作为对象字段等的方法。顺便说一句，我想到了毒丸图案……

【问题讨论】：

你的要求能再清楚一点吗？应该等待什么？ stop() 调用的客户？它应该等待什么，队列耗尽？无论如何，您是否希望调用 stop() 来阻止直到东西被耗尽？

标签： java multithreading concurrency message-queue

【解决方案1】：

您是在谈论优雅关闭您的应用程序应该执行以下操作吗？

等待队列耗尽，或者
如果排空时间过长则超时

我可能需要了解您如何排空队列，但如果您能够以可中断的方式做到这一点，您可以在获取 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 将尝试在任何当前正在运行的线程上调用中断）。

【讨论】：