Класс Flow

public final class Flow extends Object

Связанные интерфейсы и статические методы для создания компонентов с управлением потоком, в которых Publishers создают элементы, потребляемые одним или несколькими Subscribers, каждый из которых управляется объектом Subscription.

Эти интерфейсы соответствуют спецификации reactive-streams. Они применяются как в параллельных, так и в распределённых асинхронных средах: все (семь) методов определены в стиле сообщений «однонаправленного» обмена void. Взаимодействие основано на простой форме управления потоком (методе Flow.Subscription.request(long)), который можно использовать, чтобы избежать проблем с управлением ресурсами, которые иначе могут возникнуть в системах на основе «push»-модели.

Примеры. Flow.Publisher обычно определяет собственную реализацию Flow.Subscription, создаёт её в методе subscribe и передаёт вызывающему коду Flow.Subscriber. Он асинхронно публикует элементы для подписчика, обычно с помощью Executor. Например, ниже приведён очень простой издатель, который выдаёт (по запросу) единственный элемент TRUE одному подписчику. Поскольку подписчик получает только один элемент, этот класс не использует буферизацию и управление порядком, необходимые в большинстве реализаций (например, SubmissionPublisher), и опускает некоторую обработку ошибок, необходимую для полного соответствия спецификации Reactive Streams.

class OneShotPublisher implements Publisher<Boolean> {
  private final ExecutorService executor = ForkJoinPool.commonPool(); // daemon-based
  private boolean subscribed; // true after first subscribe
  public synchronized void subscribe(Subscriber<? super Boolean> subscriber) {
    if (subscribed)
      subscriber.onError(new IllegalStateException()); // only one allowed
    else {
      subscribed = true;
      subscriber.onSubscribe(new OneShotSubscription(subscriber, executor));
    }
  }
  static class OneShotSubscription implements Subscription {
    private final Subscriber<? super Boolean> subscriber;
    private final ExecutorService executor;
    private Future<?> future; // to allow cancellation
    private boolean completed;
    OneShotSubscription(Subscriber<? super Boolean> subscriber,
                        ExecutorService executor) {
      this.subscriber = subscriber;
      this.executor = executor;
    }
    public synchronized void request(long n) {
      if (!completed) {
        completed = true;
        if (n <= 0) {
          IllegalArgumentException ex = new IllegalArgumentException();
          executor.execute(() -> subscriber.onError(ex));
        } else {
          future = executor.submit(() -> {
            subscriber.onNext(Boolean.TRUE);
            subscriber.onComplete();
          });
        }
      }
    }
    public synchronized void cancel() {
      completed = true;
      if (future != null) future.cancel(false);
    }
  }
}

Flow.Subscriber организует запрос и обработку элементов. Элементы (вызовы Flow.Subscriber.onNext(T)) не отправляются, если их не запросили, но можно запросить несколько элементов. Многие реализации Subscriber могут организовать это так, как показано в следующем примере: размер буфера 1 обеспечивает пошаговую обработку, а большие размеры обычно позволяют эффективнее перекрывать обработку и уменьшить количество взаимодействий; например, значение 64 поддерживает общее число невыполненных запросов в диапазоне от 32 до 64. Поскольку вызовы методов Subscriber для данного объекта Flow.Subscription строго упорядочены, этим методам не нужно использовать блокировки или volatile-переменные, если только Subscriber не управляет несколькими подписками (в этом случае лучше определить несколько Subscriber, каждый со своей подпиской).

class SampleSubscriber<T> implements Subscriber<T> {
  final Consumer<? super T> consumer;
  Subscription subscription;
  final long bufferSize;
  long count;
  SampleSubscriber(long bufferSize, Consumer<? super T> consumer) {
    this.bufferSize = bufferSize;
    this.consumer = consumer;
  }
  public void onSubscribe(Subscription subscription) {
    long initialRequestSize = bufferSize;
    count = bufferSize - bufferSize / 2; // re-request when half consumed
    (this.subscription = subscription).request(initialRequestSize);
  }
  public void onNext(T item) {
    if (--count <= 0)
      subscription.request(count = bufferSize - bufferSize / 2);
    consumer.accept(item);
  }
  public void onError(Throwable ex) { ex.printStackTrace(); }
  public void onComplete() {}
}

Значение по умолчанию defaultBufferSize() может стать полезной отправной точкой при выборе размеров запросов и буферов компонентов Flow с учётом ожидаемой скорости, ресурсов и сценариев использования. Если же управление потоком никогда не требуется, подписчик может изначально запросить практически неограниченное количество элементов, как показано ниже:

class UnboundedSubscriber<T> implements Subscriber<T> {
  public void onSubscribe(Subscription subscription) {
    subscription.request(Long.MAX_VALUE); // effectively unbounded
  }
  public void onNext(T item) { use(item); }
  public void onError(Throwable ex) { ex.printStackTrace(); }
  public void onComplete() {}
  void use(T item) { ... }
}