java - RxJava 一次只执行一个 Observable

Question

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

1. 搜索事件被转换为数据流。

2. 在每次新搜索时，我们都需要启动异步网络操作并关闭之前的搜索。这个效果怎么存档？

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);

Answer 1

将 PublishSubject 与 switchMap 一起使用:

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

要充分利用此 networkOperationObservable 将明智地响应 unsubscribe 调用(如关闭 Socket 或其他)。 Observable.using 是通常选择的工具。