java - 防止多个异步调用同时进行而不阻塞

Question

这基本上是我的问题:

while (true) {
  if (previous 'doWorkAsync' method is not still in flight) {
    doWorkAsync()  // this returns immediately
  }
  wait set amount of time
}

我想到了几个解决方案:

1. 阻塞直到 doWorkAsync() 完成。由于一些原因，这对我来说是不可取的。它(可能)导致等待时间比我在“等待一些设定时间”行中真正需要的时间更长(例如，如果 doWorkAsync 需要 5 秒，而设定的等待时间为 10 秒，这将导致 15 秒在通话之间等待，这不是我想要的)。当然，我可以通过减少等待时间来解释这一点，但不知怎的，它就是感觉笨拙。它还会不必要地占用该线程。该线程无需等待此任务返回，而是可以处理其他工作，例如进行配置更新，以便下一次调用 doWorkAsync() 时获得新数据。

2. 使用门控机制。想到的最简单的实现是 boolean 值，在调用 doWorkAsync() 之前设置，并在 doWorkAsync() 完成时取消设置。这基本上就是我现在正在做的，但我不确定它是否是一种反模式？？

#2 是正确的方法吗，还是有更好的方法来解决这个问题？

编辑:如果有帮助，doWorkAsync() 会返回一个 ListenableFuture( Guava )。

原始问题可能不是 100% 清楚。这是症结所在。如果异步请求在给定超时之前完成，则此代码将始终有效。但是，如果异步任务需要 SET_AMOUNT_OF_TIME + epsilon 才能完成，那么这段代码将根据需要 hibernate 两倍的时间，这是我试图避免的。

Answer 1

最简单的方法是使用 Java 中已有的 wait 和 notifyAll 方法。您需要做的就是使用一个 AtomicBoolean 作为标志并阻止它，直到另一个 Thread 告诉您发生了一些变化。

这与您的方法之间的区别在于阻塞线程不执行任何操作，而轮询线程使用 CPU 时间。

这是一个使用两个 Thread 的简单示例 - Runnable “First” 已提交并等待 done 直到 Runnable “Second” 通知它已更改标志。

public class App {

    private static final AtomicBoolean done = new AtomicBoolean(false);

    private static final class First implements Runnable {

        @Override
        public void run() {
            while (!done.get()) {
                System.out.println("Waiting.");
                synchronized (done) {
                    try {
                        done.wait();
                    } catch (InterruptedException ex) {
                        return;
                    }
                }
            }
            System.out.println("Done!");
        }
    }

    private static final class Second implements Runnable {

        @Override
        public void run() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                return;
            }
            done.set(true);
            synchronized (done) {
                done.notifyAll();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final ExecutorService executorService = Executors.newFixedThreadPool(2);

        executorService.submit(new First());
        Thread.sleep(1000);
        executorService.submit(new Second());
        executorService.shutdown();

    }
}

sleep 调用只是为了表明可以执行任意长度的任务，显然它们不是必需的。

需要注意的是，First 每次 进入循环时都会打印“waiting”，如果您运行该代码，它只会打印一次。第二点要注意的是 First 会立即对标志的更改使用react，因为它被告知在标志更改时唤醒并重新检查。

我在 InterruptedException block 中使用了 return，您可能想使用 Thread.currentThread().interrupt() 来代替，这样如果进程被虚假中断，该进程不会终止。

更高级的方法是使用Lock 和Condition

public class App {

    private static final Lock lock = new ReentrantLock();
    private static final Condition condition = lock.newCondition();

    private static final class First implements Runnable {

        @Override
        public void run() {
            lock.lock();
            System.out.println("Waiting");
            try {
                condition.await();
            } catch (InterruptedException ex) {
                return;
            } finally {
                lock.unlock();
            }
            System.out.println("Done!");
        }
    }

    private static final class Second implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                Thread.sleep(1000);
                condition.signalAll();
            } catch (InterruptedException ex) {
                return;
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final ExecutorService executorService = Executors.newFixedThreadPool(2);

        executorService.submit(new First());
        Thread.sleep(1000);
        executorService.submit(new Second());
        executorService.shutdown();

    }
}

在这种情况下，首先 获取Lock 对象上的锁，然后立即调用Condition 上的await。释放锁并阻止 Condition。

Second 然后获取 Lock 上的锁，并在唤醒 的 Condition 上调用 signalAll首先。

首先 然后重新获取锁并继续执行，打印“完成!”。

编辑

OP 希望在指定的时间段内调用方法 doWorkAsync，如果该方法花费的时间少于该时间段，则进程必须等待。如果该方法花费的时间更长，则应在之后立即再次调用该方法。

任务需要在一定时间后停止。

该方法在任何时候都不应同时运行多次。

最简单的方法是从 ScheduledExecutorService 调用方法，Runnable 将封装该方法并在 上调用 get >Future - 阻塞预定的执行器直到它完成。

这保证调用方法时至少 WAIT_TIME_BETWEEN_CALLS_SECS 延迟。

然后安排另一个任务在设定的时间后杀死第一个任务。

final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
final Future<?> taskHandle = scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        final ListenableFuture<Void> lf = doWorkAsync();
        try {
            doWorkAsync().get();
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        } catch (ExecutionException ex) {
            throw new RuntimeException(ex);
        }
    }
}, 0, WAIT_TIME_BETWEEN_CALLS_SECS, TimeUnit.SECONDS);
scheduledExecutorService.schedule(new Runnable() {
    @Override
    public void run() {
        taskHandle.cancel(false);
    }
}, TOTAL_TIME_SECS, TimeUnit.SECONDS);

最好的解决方案是在 ScheduledExecutorService 上调用原始 Runnable，而不是在另一个执行器上调用它并在 ListenableFuture 上阻塞。