Flow 类
定义
重要
一些信息与预发行产品相关,相应产品在发行之前可能会进行重大修改。 对于此处提供的信息,Microsoft 不作任何明示或暗示的担保。
用于建立流控制组件的相互关联接口和静态方法,其中 Publisher Publishers
生成由一个或多个 Subscriber
Subscribers
由 a 管理 Subscription
Subscription
的项目。
[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
- 继承
- 属性
注解
用于建立流控制组件的相互关联接口和静态方法,其中 Publisher Publishers
生成由一个或多个 Subscriber Subscribers
由 a 管理 Subscription Subscription
的项目。
这些接口对应于 反应式流 规范。 它们同时适用于并发和分布式异步设置:所有(7)方法都以 void
“单向”消息样式定义。 通信依赖于一种简单的流控制(方法 Subscription#request
)形式,可用于避免在基于“推送”的系统中可能出现的资源管理问题。
<b>示例。</b> A Publisher
通常定义自己的 Subscription
实现;在方法 subscribe
中构造一个,并将其颁发给调用 Subscriber
。 它以异步方式将项发布到订阅服务器,通常使用 Executor
. 例如,下面是一个非常简单的发布者,仅向单个订阅者颁发单个 TRUE
项(请求时)。 由于订阅服务器只接收单个项,因此此类不使用大多数实现中所需的缓冲和排序控制。
{@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 文档
本页的某些部分是根据 Android 开放源代码项目创建和共享的工作进行的修改,并根据 Creative Commons 2.5 属性许可证中所述的术语使用。
属性
Class |
返回此 |
Handle |
基础 Android 实例的句柄。 (继承自 Object) |
JniIdentityHashCode |
用于建立流控制组件的相互关联接口和静态方法,其中 |
JniPeerMembers |
用于建立流控制组件的相互关联接口和静态方法,其中 |
PeerReference |
用于建立流控制组件的相互关联接口和静态方法,其中 |
ThresholdClass |
此 API 支持 Mono for Android 基础结构,不打算直接从代码使用。 (继承自 Object) |
ThresholdType |
此 API 支持 Mono for Android 基础结构,不打算直接从代码使用。 (继承自 Object) |
方法
Clone() |
创建并返回此对象的副本。 (继承自 Object) |
DefaultBufferSize() |
返回发布服务器或订阅服务器缓冲的默认值,该值可能在缺少其他约束时使用。 |
Dispose() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
Dispose(Boolean) |
用于建立流控制组件的相互关联接口和静态方法,其中 |
Equals(Object) |
指示其他对象是否“等于”此对象。 (继承自 Object) |
GetHashCode() |
返回对象的哈希代码值。 (继承自 Object) |
JavaFinalize() |
当垃圾回收确定不再引用该对象时,由对象上的垃圾回收器调用。 (继承自 Object) |
Notify() |
唤醒正在等待此对象的监视器的单个线程。 (继承自 Object) |
NotifyAll() |
唤醒正在等待此对象的监视器的所有线程。 (继承自 Object) |
SetHandle(IntPtr, JniHandleOwnership) |
设置 Handle 属性。 (继承自 Object) |
ToArray<T>() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
ToString() |
返回对象的字符串表示形式。 (继承自 Object) |
UnregisterFromRuntime() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
Wait() |
使当前线程等待,直到唤醒它,通常是通过 em 通知/em> 或 <em>interrupted</em>。<>< (继承自 Object) |
Wait(Int64) |
使当前线程等待直到唤醒,通常是通过 <em>通知</em> 或 <em interrupted</em>>,或直到经过一定数量的实时。 (继承自 Object) |
Wait(Int64, Int32) |
使当前线程等待直到唤醒,通常是通过 <em>通知</em> 或 <em interrupted</em>>,或直到经过一定数量的实时。 (继承自 Object) |
显式接口实现
IJavaPeerable.Disposed() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.DisposeUnlessReferenced() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.Finalized() |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.JniManagedPeerState |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.SetJniIdentityHashCode(Int32) |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.SetJniManagedPeerState(JniManagedPeerStates) |
用于建立流控制组件的相互关联接口和静态方法,其中 |
IJavaPeerable.SetPeerReference(JniObjectReference) |
用于建立流控制组件的相互关联接口和静态方法,其中 |
扩展方法
JavaCast<TResult>(IJavaObject) |
执行 Android 运行时检查的类型转换。 |
JavaCast<TResult>(IJavaObject) |
用于建立流控制组件的相互关联接口和静态方法,其中 |
GetJniTypeName(IJavaPeerable) |
用于建立流控制组件的相互关联接口和静态方法,其中 |