当周期性任务被取消并完成时回调

Question

我有两个任务:第一个任务(work)正在重复发生，第二个任务(cleanup)正在释放一些资源。cleanup任务应该在重复出现的work任务完成后正好运行一次，并且不能再运行。

我的第一反应是这样的:

ScheduledExecutorService service = ...;
ScheduledFuture<?> future = service.scheduleAtFixedRate(work, ...);
// other stuff happens
future.cancel(false);
cleanup.run();

这里的问题是cancel()立即返回。因此，如果work正在运行，那么cleanup将与它重叠。

理想情况下，我会使用Guava的Futures.addCallback(ListenableFuture future, FutureCallback callback)。(Guava 15可能有类似的东西)。

同时，当future被取消和 work不再运行时，如何触发回调?

Answer 1

这是我想到的解决方案。这似乎非常简单，但我仍然认为存在更常见和/或更优雅的解决方案。我真的很想在像Guava这样的图书馆里看到一个……

首先，我创建了一个包装器来对我的Runnables强制互斥:

private static final class SynchronizedRunnable implements Runnable {
    private final Object monitor;
    private final Runnable delegate;
    private SynchronizedRunnable(Object monitor, Runnable delegate) {
        this.monitor = monitor;
        this.delegate = delegate;
    }
    @Override
    public void run() {
        synchronized (monitor) {
            delegate.run();
        }
    }
}

然后创建一个包装器，在成功调用cancel时触发回调:

private static final class FutureWithCancelCallback<V> extends ForwardingFuture.SimpleForwardingFuture<V> {
    private final Runnable callback;
    private FutureWithCancelCallback(Future<V> delegate, Runnable callback) {
        super(delegate);
        this.callback = callback;
    }
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
            boolean cancelled = super.cancel(mayInterruptIfRunning);
            if (cancelled) {
                callback.run();
            }
            return cancelled;
    }
}

然后我用我自己的方法把它们组合在一起:

private Future<?> scheduleWithFixedDelayAndCallback(ScheduledExecutorService service, Runnable work, long initialDelay, long delay, TimeUnit unit, Runnable cleanup) {
    Object monitor = new Object();
    Runnable monitoredWork = new SynchronizedRunnable(monitor, work);
    Runnable monitoredCleanup = new SynchronizedRunnable(monitor, cleanup);
    Future<?> rawFuture = service.scheduleAtFixedRate(monitoredWork, initialDelay, delay, unit);
    Future<?> wrappedFuture = new FutureWithCancelCallback(rawFuture, monitoredCleanup);
    return wrappedFuture;
}

Answer 2

我再试一次。您可以增强命令，也可以包装已执行的Runnable/Callable。看这个:

public static class RunnableWrapper implements Runnable {
    private final Runnable original;
    private final Lock lock = new ReentrantLock();
    public RunnableWrapper(Runnable original) {
        this.original = original;
    }
    public void run() {
        lock.lock();
        try {
            this.original.run();
        } finally {
            lock.unlock();
        }
    }
    public void awaitTermination() {
        lock.lock();
        try {
        } finally {
            lock.unlock();
        }
    }
}

所以你可以把你的代码改成

ScheduledExecutorService service = ...;
RunnableWrapper wrapper = new RunnableWrapper(work);
ScheduledFuture<?> future = service.scheduleAtFixedRate(wrapper, ...);
// other stuff happens
future.cancel(false);
wrapper.awaitTermination();
cleanup.run();

调用cancel后，要么work不再运行，awaitTermination()立即返回，要么它正在运行，awaitTermination()阻塞，直到完成。

Answer 3

为什么不呢

// other stuff happens
future.cancel(false);
service.shutdown();
service.awaitTermination(1, TimeUnit.DAYS);
cleanup.run();

这将告诉您的执行器服务关闭，从而允许您等待可能正在运行的work完成。