java - RxJava一次只能执行一个Observable

我们有带有搜索字段的窗口。每次用户输入内容时都会执行搜索。


搜索事件转换为数据流。
在每个新搜索中，我们需要启动异步网络操作并关闭上一个。如何存档此效果？


Edit1：这是我尝试过的。它执行所有可观察的东西！不是唯一的最后一个，哪里有错误？

PublishSubject<Integer> subject = PublishSubject.create();

subject.switchMap(integer -> Observable.fromCallable(() -> {
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "-" + integer + "-";
})).subscribe(s -> System.out.print(s));
for (int i = 0; i < 30; i++) subject.onNext(i);

最佳答案

将PublishSubject与switchMap一起使用：

PublishSubject<String> subject = PublishSubject.create();
subject
  .switchMap(
    x -> networkOperationObservable(x)
           .subscribeOn(Schedulers.io))
  .subscribe(subscriber);

充分利用此networkOperationObservable可以对unsubscribe调用做出明智的响应（如关闭Socket或类似命令）。 Observable.using通常是为此选择的工具。

关于java - RxJava一次只能执行一个Observable，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/40425057/