我想将订户与Reactor Flux连接起来。但是我的小例子没有产生任何输出:
public static void main(String[] args) throws InterruptedException {
Flux.just("a", "b", "c")
.subscribe(new BaseSubscriber<String>() {
@Override
protected void hookOnSubscribe(Subscription ignored) {
}
@Override
protected void hookOnNext(String value) {
System.out.print(value);
}
});
Thread.sleep(1000);
}
我用org.reactivestreams.Subscriber尝试了相同的方法:
Flux.just("a", "b", "c")
.subscribe(new Subscriber<String>() {
@Override
public void onSubscribe(Subscription s) {
}
@Override
public void onNext(String s) {
System.out.println(s);
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
}
});
但是同样,没有输出。
如果我尝试以下最简单的方法:
Flux.just("a", "b", "c").subscribe(System.out::println);
...产生预期的输出:
a
b
c
如何使用订阅服务器从Flux接收值?
最佳答案
关键是要从源中另外request
n个值。在此之前,没有任何反应。BaseSubscriber
的示例:
Flux<String> source = Flux.just("a", "b", "c");
source.subscribe(new BaseSubscriber<String>() {
@Override
protected void hookOnSubscribe(Subscription subscription) {
request(1); // <-- here
}
@Override
protected void hookOnNext(String value) {
request(1); // <-- here
System.out.println(value);
}
});
与标准
Subscriber
相同:Flux<String> source = Flux.just("a", "b", "c");
source.subscribe(new Subscriber<String>() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1); // <-- here
}
@Override
public void onNext(String s) {
subscription.request(1); // <-- here
System.out.println(s);
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
}
});