Flow クラス

定義

[Android.Runtime.Register("java/util/concurrent/Flow", ApiSince=30, DoNotGenerateAcw=true)]
public sealed class Flow : Java.Lang.Object

[<Android.Runtime.Register("java/util/concurrent/Flow", ApiSince=30, DoNotGenerateAcw=true)>]
type Flow = class
    inherit Object

継承

Object

JavaObject

Object

Flow

属性

RegisterAttribute

注釈

Publisher Publishersが 1 つ以上のSubscriber Subscribersによって消費される項目を生成するフロー制御コンポーネントを確立するための相互に関連するインターフェイスと静的メソッド。各コンポーネントは、Subscription Subscriptionによって管理されます。

これらのインターフェイスは、 リアクティブ ストリーム の仕様に対応します。 これらは、同時実行と分散の両方の非同期設定に適用されます。すべての (7 つの) メソッドは、"一方向" メッセージ スタイル void 定義されます。 通信は、"プッシュ" ベースのシステムで発生する可能性があるリソース管理の問題を回避するために使用できる単純な形式のフロー制御 (メソッド Subscription#request) に依存します。

<b>Examples.</b> A Publisher は、通常、独自の Subscription 実装を定義します。メソッド subscribe で 1 つを構築し、それを呼び出し元の Subscriberに発行します。 通常は、 Executorを使用して、アイテムをサブスクライバーに非同期的に発行します。 たとえば、1 つのサブスクライバーに対して 1 つの TRUE 項目のみを (要求された場合に) 発行する非常に単純な発行元を次に示します。 サブスクライバーは 1 つの項目のみを受け取るため、このクラスでは、ほとんどの実装で必要なバッファリングと順序付けコントロールは使用されません。

{@code
            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);
                }
              }
            }}

Subscriberは、アイテムの要求と処理を手配します。 項目 ( Subscriber#onNextの呼び出し) は、要求されない限り発行されませんが、複数の項目が要求される場合があります。 多くのサブスクライバー実装では、次の例のスタイルでこれを配置できます。バッファー サイズが 1 つの単一ステップであり、サイズが大きいほど、通常、通信の少ない効率的な重複処理が可能になります。たとえば、値が 64 の場合、未処理の要求の合計数は 32 から 64 の間に保持されます。 特定の Subscription に対するサブスクライバー メソッドの呼び出しは厳密に順序付けられているため、サブスクライバーが複数のサブスクリプションを保持しない限り、これらのメソッドでロックまたは揮発性を使用する必要はありません (その場合は、複数のサブスクライバーを定義し、それぞれが独自のサブスクリプションを持つ方がよい)。

{@code
            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 コンポーネントの要求サイズと容量を選択するための便利な開始点となる場合があります。 または、フロー制御が不要な場合、サブスクライバーは最初に、次のように実質的に無制限の数の項目を要求する場合があります。

{@code
            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) { ... }
            }}

9 に追加されました。

java.util.concurrent.Flowの Java ドキュメント。

このページの一部は、によって作成および共有され、に記載されている条件に従って使用される作業に基づく変更です。