Java "in-order"信号量

Question

我有一个问题，并且对如何解决它有一个模糊的想法，但我会尝试过度共享上下文以避免 XY problem .

我有一个异步方法，它立即返回 guava ListenableFuture我需要调用数十万或数百万次( future 本身可能需要一段时间才能完成)。我无法真正改变该方法的内部结构。内部涉及一些严重的资源争用，因此我想限制同时发生的对该方法的调用次数。所以我尝试使用信号量:

public class ConcurrentCallsLimiter<In, Out> {
  private final Function<In, ListenableFuture<Out>> fn;
  private final Semaphore semaphore;
  private final Executor releasingExecutor;

  public ConcurrentCallsLimiter(
      int limit, Executor releasingExecutor,
      Function<In, ListenableFuture<Out>> fn) {
    this.semaphore = new Semaphore(limit);
    this.fn = fn;
    this.releasingExecutor = releasingExecutor;
  }

  public ListenableFuture<Out> apply(In in) throws InterruptedException {
    semaphore.acquire();
    ListenableFuture<Out> result = fn.apply(in);
    result.addListener(() -> semaphore.release(), releasingExecutor);
    return result;
  }
}

那么我就可以将我的调用包装在这个类中并调用它:

ConcurrentLimiter<Foo, Bar> cl =
    new ConcurrentLimiter(10, executor, someService::turnFooIntoBar);
for (Foo foo : foos) {
  ListenableFuture<Bar> bar = cl.apply(foo);
  // handle bar (no pun intended)
}

这有点的作用。问题是尾部延迟非常糟糕。有些调用“不走运”，最终需要花费很长时间来尝试获取该方法调用中的资源。一些内部指数退避逻辑加剧了这种情况，与更加渴望并在重试之前等待更短时间的新调用相比，不幸的调用获得所需资源的机会越来越少。

解决这个问题的理想方法是，如果有一个类似于该信号量但具有顺序概念的东西。例如，如果限制为 10，则第 11 个调用当前必须等待前 10 个调用中的任何一个调用完成。我希望第 11 个调用必须等待第一个调用完成。这样，“不幸”的调用就不会因为不断打进来的新调用而继续匮乏。

似乎我可以为调用分配一个整数序列号，并以某种方式跟踪尚未完成的最低序列号，但不太明白如何实现这一点，特别是因为 AtomicInteger 或其他方面实际上没有任何有用的“等待”方法。

Answer 1

您可以使用公平参数创建信号量:

Semaphore(int permits, boolean fair)

//fair - 如果此信号量保证先进先出授予争用许可证，则为 true，否则为 false