如何在 Java 9 中实现订阅者接口?
javaobject oriented programmingprogramming更新于 2025/10/11 18:52:17
Java 9 通过引入几个接口支持创建响应式流:Publisher、Subscriber、Subscription和实现Publisher 接口的SubmissionPublisher 类。每个接口可以扮演不同的角色，符合响应式流的原则。
我们可以使用Subscriber 接口来订阅发布者正在发布的数据。我们需要实现 Subscriber 接口，并为抽象方法提供实现。
Flow.Subscriber 接口方法:
- onComplete():当 Publisher 对象完成其角色时，会调用此方法。
- onError():当 Publisher 出现问题并通知给订阅者时，会调用此方法。
- onNext():每当 Publisher 有新信息需要通知所有订阅者时，会调用此方法。
- onSubscribe():当 Publisher 添加订阅者时，会调用此方法。
示例
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;
public class SubscriberImplTest {
public static class Subscriber implements Flow.Subscriber<Integer> {
private Flow.Subscription subscription;
private boolean isDone;
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println("Subscribed");
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Integer item) {
System.out.println("Processing " + item);
this.subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("Processing done");
isDone = true;
}
}
public static void main(String args[]) throws InterruptedException {
SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
Subscriber subscriber = new Subscriber();
publisher.subscribe(subscriber);
IntStream intData = IntStream.rangeClosed(1, 10);
intData.forEach(publisher::submit);
publisher.close();
while(!subscriber.isDone) {
Thread.sleep(10);
}
System.out.println("Done");
}
}
输出
Subscribed Processing 1 Processing 2 Processing 3 Processing 4 Processing 5 Processing 6 Processing 7 Processing 8 Processing 9 Processing 10 Processing done Done
相关文章
有用资源
java 参考教程 - 该教程包含有关 java 的更多信息:https://www.cainiaomax.com/java/
