Flow API (java.util.concurrent.Flow) 已在 Java 9 中引入。它有助于了解发布商和S订阅者界面交互以执行所需操作的不同方式。
Flow API 由发布者、订阅者、订阅和处理器接口组成，这些接口可以基于反应式流规范。
在下面的示例中，我们可以使用发布者-订阅者接口来实现 Flow API。
示例
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
public class FlowAPITest {
public static void main(String args[]) {
<strong>Publisher<Integer></strong> publisherSync = new <strong>Publisher<Integer></strong>() { <strong>// Create publisher</strong>
<strong>@Override</strong>
public void <strong>subscribe</strong>(Subscriber<? super Integer><!--? super Integer--> subscriber) {
for(int i = 0; i < 10; i++) {
System.out.println(Thread.currentThread().getName() + " | Publishing = " + i);
subscriber.<strong>onNext</strong>(i);
}
subscriber.<strong>onComplete</strong>();
}
};
<strong>Subscriber<Integer></strong> subscriberSync = new <strong>Subscriber<Integer></strong>() { <strong>// Create subscriber</strong>
<strong>@Override</strong>
public void <strong>onSubscribe</strong>(Subscription subscription) {
}
<strong>@Override</strong>
public void <strong>onNext</strong>(Integer item) {
System.out.println(Thread.currentThread().getName() + " | Received = " + item);
try {
Thread.sleep(100);
} catch(InterruptedException e) {
e.printStackTrace();
}
}
<strong>@Override</strong>
public void <strong>onError</strong>(Throwable throwable) {
}
<strong>@Override</strong>
public void <strong>onComplete()</strong> {
}
};
publisherSync.<strong>subscribe</strong>(subscriberSync);
}
}输出
<strong>main | Publishing = 0 main | Received = 0 main | Publishing = 1 main | Received = 1 main | Publishing = 2 main | Received = 2 main | Publishing = 3 main | Received = 3 main | Publishing = 4 main | Received = 4 main | Publishing = 5 main | Received = 5 main | Publishing = 6 main | Received = 6 main | Publishing = 7 main | Received = 7 main | Publishing = 8 main | Received = 8 main | Publishing = 9 main | Received = 9</strong>
