<p>之前看了 <a href="/misc/goto?guid=4959738217635070674" rel="nofollow,noindex">拉丁吴</a> 关于 RxJava 的一些文章，获益匪浅可惜的是大佬并没有对于各个运算符进行介绍，关于 RxJava2 也只给了一个 GitHub 上的 Demo <a href="/misc/goto?guid=4959738217726985527" rel="nofollow,noindex">RxJava2-Android-Samples</a> ，这个工程里面的注释不多，只能自己一个一个去试了，这种小例子感觉还是直接用 Java 来的方便些，于是自己就写了一个 <a href="/misc/goto?guid=4959738217813346560" rel="nofollow,noindex">RxJava2Deme</a></p>    <p>这篇文章就是对这个例子的一个一个操作进行了简单的说明，仅此而已为了简化下面的代码，先定义了一些复用的函数，有点多，就不贴上来了，大概就是 Observer<Integer> getObserver(int n) 这样的。 <a href="/misc/goto?guid=4959738217899665375" rel="nofollow,noindex">文件地址</a></p>    <h2>操作符</h2>    <h3>concat()</h3>    <p><a href="/misc/goto?guid=4959649645906935883" rel="nofollow,noindex">doc</a></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.concat(Observable.just(1, 2, 3), Observable.just(5, 6, 7))          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:1  0 onNext:2  0 onNext:3  0 onNext:5  0 onNext:6  0 onNext:7  0 onComplete     </code></pre>    <p>很简单的连接操作</p>    <h3>distinct()</h3>    <p><a href="/misc/goto?guid=4959738218017201762" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/548106706d934af050f67aef7d03bc36.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 1, 2, 3)          .distinct()          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onNext:2  0 onNext:3  0 onComplete     </code></pre>    <p>分析：就是很简单的去重</p>    <h3>filter()</h3>    <p><a href="/misc/goto?guid=4959657463019796329" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/8201a6c7a884f07566e631433950f203.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 3, 4)          .filter(i -> i > 2)          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:3  0 onNext:4  0 onComplete     </code></pre>    <p>分析：筛选也没啥说的，大家都会了</p>    <h3>buffer()</h3>    <p><a href="/misc/goto?guid=4959738218132423327" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/c2a7eda9f21b13dc1d37193839882b8a.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable<Integer> observable = Observable.just(1, 2 ,3, 4, 5);  observable.buffer(3, 2).subscribe(getListObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0list onSubscribe  0list onNext:[1, 2, 3]  0list onNext:[3, 4, 5]  0list onNext:[5]  0list onComplete     </code></pre>    <p>解析：</p>    <pre>  <code class="language-java">buffer(count, skip)` 从定义就差不多能看出作用了，将 observable 中的数据按 skip（步长）分成最长不超过 count 的 buffer，然后生成一个 observable     </code></pre>    <h3>debounce()</h3>    <p><a href="/misc/goto?guid=4959647703242238889" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/2418495b0a24fef0c7082ba9adcf3a31.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.create((ObservableOnSubscribe<Integer>) e -> {      e.onNext(1);      Thread.sleep(1000);      e.onNext(2);      Thread.sleep(400);      e.onNext(3);      Thread.sleep(1000);      e.onNext(4);      Thread.sleep(400);      e.onNext(5);      Thread.sleep(400);      e.onNext(6);      Thread.sleep(1000);      e.onComplete();  }).subscribeOn(Schedulers.newThread())          .observeOn(Schedulers.newThread())          .debounce(500, TimeUnit.MILLISECONDS)          .subscribe(getObserver(0));  Thread.sleep(10000);     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:1  0 onNext:3  0 onNext:6  0 onComplete     </code></pre>    <p>分析：</p>    <p>这么理解，在一段时间内只能有一次 onNext()</p>    <p>看上面的例子，我们将时间间隔设置为了 500ms</p>    <p>onNext(1) -> sleep(1000) （因为 1000 > 500) 所以 onNext(1) 成功执行 -> onNext(2) -> sleep(400) （因为 400 < 1000 所以 onNext(2) 被除掉了</p>    <p>这样依次下去，需要注意的只要写了 onNext() 时间就会被重置，即使这个 onNext() 被除掉了</p>    <p>在上面这个例子中 4 和 5 隔了 400ms，所以 4 没有触发，但是时间得重新计算</p>    <h3>defer()</h3>    <p><a href="/misc/goto?guid=4959738218238621705" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/c38c7e44f4ff4580ef593f264dcb8d52.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {      @Override      public ObservableSource<Integer> call() throws Exception {          return Observable.just(0, 1, 2, 3);      }  });        observable.subscribe(getObserver(0));  observable.subscribe(getObserver(1));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onNext:2  0 onNext:3  0 onComplete  1 onSubscribe  1 onNext:0  1 onNext:1  1 onNext:2  1 onNext:3  1 onComplete     </code></pre>    <p>分析：</p>    <p>就是在每次订阅的时候就会创建一个新的 Observable</p>    <h3>interval()</h3>    <p><a href="/misc/goto?guid=4959738218332401708" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/905d49717fda6b267addd5c6ed16f5c7.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.interval(3, 2, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {      @Override      public void accept(Long aLong) throws Exception {          p("accept " + aLong + " at " + System.currentTimeMillis());      }  });     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">accept 0 at 1487498704028  accept 1 at 1487498706028  accept 2 at 1487498708028  accept 3 at 1487498710028  accept 4 at 1487498712028  ...     </code></pre>    <p>分析：</p>    <p>类似于一个定时任务这样吧，也可以像下面这么写：</p>    <pre>  <code class="language-java">Observable.interval(3, 2, TimeUtil.SECONDS);  Observable.just(1, 2, 3).subscribeWith(getDisposableObserver(1));     </code></pre>    <h3>last()</h3>    <p><a href="/misc/goto?guid=4959738218414980893" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/e9ec56e4bb8d1f4cfc072a088c425d52.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 2).last(3).subscribe(getSingleObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onSuccess :2     </code></pre>    <p>分析：</p>    <p>就是取出最后一个值，参数是没有值的时候的默认值，比如这样：</p>    <pre>  <code class="language-java">Observable.create((ObservableOnSubscribe<Integer>) Emitter::onComplete).last(3)          .subscribe(getSingleObserver(0));     </code></pre>    <p>输出就是 3 了</p>    <p>另见 lastOrError() 方法，区别就是 lastOrError() 没有默认值，没值就触发错误</p>    <h3>map()</h3>    <p><a href="/misc/goto?guid=4959631851652662337" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/8333a164ba7b155bc0cb4fd902c4cb00.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 2, 3)          .map(i -> "string" + i)          .subscribe(getStringObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:string0  0 onNext:string1  0 onNext:string2  0 onNext:string3  0 onComplete     </code></pre>    <p>分析：</p>    <p>把一个 Observable 转成另一个 Observable</p>    <p>###</p>    <p><a href="/misc/goto?guid=4959738218523253816" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/81851b6b9c689b74795c7e9dfbf64059.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.merge(Observable.just(0, 1), Observable.just(3, 4))          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onNext:3  0 onNext:4  0 onComplete     </code></pre>    <p>分析：</p>    <p>将多个 Observable 合起来，例子中只是两个</p>    <p>参数也支持使用迭代器将更多的组合起来</p>    <p>###</p>    <p><a href="/misc/goto?guid=4959738218621173659" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/f24a081c09ef53c23f4528b5dd847f97.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(1, 2, 3)          .reduce((i1, i2) -> i1 + i2)          .subscribe(getMaybeObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onSuccess 6     </code></pre>    <p>分析：</p>    <p>就是依次用一个方法处理每个值，可以有一个 seed 作为初始值</p>    <p>不指定 seed 则第一次传入的就是第一第二个</p>    <p>###</p>    <p><a href="/misc/goto?guid=4959738218702669093" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/d045d1b690551e1546992b94134d450b.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">PublishSubject<Integer> publishSubject = PublishSubject.create();  Observable<Integer> observable = publishSubject.replay().autoConnect();     observable.subscribe(getObserver(0));  publishSubject.onNext(0);  observable.subscribe(getObserver(1));  publishSubject.onNext(1);  publishSubject.onComplete();     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  1 onSubscribe  1 onNext:0  0 onNext:1  1 onNext:1  0 onComplete  1 onComplete     </code></pre>    <p>分析：</p>    <p>从结果可以很明显的看出来，使用了 replay() 后， subscribe(getObserver(1)) 之前的数据也被传入了</p>    <p>相对于 cache() 来说， replay() 提供了更多可控制的选项，在实际使用中可以通过指定 bufferSize 来防止内存占用过大等</p>    <p>还有一个 ReplaySubject 功能和这个应该是差不多的，下面就不单独说了，代码在 GitHub 上都有。</p>    <h3>merge()</h3>    <p><a href="/misc/goto?guid=4959657462922360507" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/58e16ac9c4a7a59b9d1de096babe8259.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 2, 3)          .scan((i1, i2) -> i1 + i2)          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onNext:3  0 onNext:6  0 onComplete     </code></pre>    <p>分析：</p>    <p>scan() 和上面提到的 reduce() 差不多</p>    <p>区别在于 reduce() 只输出最终结果，而 scan() 会将过程中的每一个结果输出</p>    <h3>skip()</h3>    <p><a href="/misc/goto?guid=4959738218827822572" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/8fe614af38414989058713ac8b56d43f.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 2, 3)          .skip(2)          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:2  0 onNext:3  0 onComplete     </code></pre>    <p>分析：</p>    <p>看名字也知道是干嘛的了，跳过一些数据，例子中跳过的是数据量，也可以跳过时间 skip(long time, TimeUnit unit)</p>    <h3>take()</h3>    <p><a href="/misc/goto?guid=4959646509260271122" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/deb2310825547b2fb88492af66080aaf.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.just(0, 1, 2, 3)          .skip(2)          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onComplete     </code></pre>    <p>分析：</p>    <p>从数据中取前几个出来</p>    <p>和 skip() 类似，参数可以指定为时间，那就是取出这段时间里的数据了</p>    <h3>sample() throttleFirst() throttleLast()</h3>    <p>这几个是相关的，一起说</p>    <p><a href="/misc/goto?guid=4959738218959715618" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/36084ef074056869b7efb47d8834d5b6.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.create((ObservableOnSubscribe<Integer>) e -> {      e.onNext(0);      Thread.sleep(200);      e.onNext(1);      Thread.sleep(600);      e.onNext(2);      Thread.sleep(300);      e.onNext(3);      Thread.sleep(1100);      e.onNext(4);      Thread.sleep(3000);      e.onComplete();  }).sample(1, TimeUnit.SECONDS) // throttleFirst 和 throttleLast 就将这里的 sample 改掉就行          .subscribe(getObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">sample:  0 onSubscribe  0 onNext:2  0 onNext:3  0 onNext:4  0 onComplete     throttleFirst:  0 onSubscribe  0 onNext:0  0 onNext:3  0 onNext:4  0 onComplete     throttleLast:  0 onSubscribe  0 onNext:2  0 onNext:3  0 onComplete     </code></pre>    <p>分析：</p>    <p>先从 sample() 开始吧，先看下官网上的定义</p>    <p>emit the most recent items emitted by an Observable within periodic time intervals</p>    <p>发出最接近周期点的事件</p>    <p>在例子中，我们使用了 1000 作为 时间间隔，随手画了张图，将就着看下</p>    <p><img src="https://simg.open-open.com/show/3d208088b508bec2605aad0db4d9c1b6.png"></p>    <p>在 A 点之前没有点，B 点之前最近的一个点是 2，C 点之前的是 3，所以输出 2 和 3。</p>    <p>对于 throttleFirst() ，它和 sample() 是相反的，只是要注意一下，图中的 0 点是被当做在 A 点之后处理的</p>    <p>而 throttleLast() 就是 sample() 换了个名字而已</p>    <h3>timer()</h3>    <p><a href="/misc/goto?guid=4959738219062530802" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/414a765397e0a3814b31389b4f6bc877.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.timer(1, TimeUnit.SECONDS)          .subscribe(getLongObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onComplete     </code></pre>    <p>分析：</p>    <p>定时任务了</p>    <h3>window()</h3>    <p><a href="/misc/goto?guid=4959738219181874493" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/83efb47adce43343fade68b3fb8d98fb.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.interval(1, TimeUnit.SECONDS)      .take(9)      .window(3, TimeUnit.SECONDS)      .subscribe(new Consumer<Observable<Long>>() {          int n = 0 ;          @Override          public void accept(Observable<Long> longObservable) throws Exception {              longObservable.subscribe(getLongObserver(n++));          }      });     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  0 onNext:1  0 onComplete  1 onSubscribe  1 onNext:2  1 onNext:3  1 onNext:4  1 onComplete  2 onSubscribe  2 onNext:5  2 onComplete     </code></pre>    <p>分析：</p>    <p>按照时间划分窗口，将数据发送给不同的 observable 。</p>    <h3>zip()</h3>    <p><a href="/misc/goto?guid=4959670499883097657" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/611485bcf4f58398d36a6f3470491331.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Observable.zip((ObservableSource<Integer>) observer -> {      observer.onNext(1);      observer.onNext(2);      observer.onNext(3);     }, (ObservableSource<String>) observer -> {      observer.onNext("str");      observer.onNext("text");      observer.onComplete();  }, (integer, s) -> integer + s).subscribe(getStringObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:1str  0 onNext:2text  0 onComplete     </code></pre>    <p>分析：</p>    <p>功能是可以将不同的观察者进行组合，并且 onNext() 是一对一对的，例子里的 3 就没有对象了（3 真是惨 =.=）</p>    <p>然后只要有一个执行了 onComplete 就会结束掉</p>    <h2>类</h2>    <h3>PublishSubject</h3>    <p><a href="/misc/goto?guid=4959738219307915783" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/f3eae6d22a4f91a43bca9a232df35d88.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">publishSubject.subscribe(getObserver(0));  publishSubject.onNext(0);  publishSubject.subscribe(getObserver(1));  publishSubject.onNext(1);  publishSubject.onComplete();     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onNext:0  1 onSubscribe  0 onNext:1  1 onNext:1  0 onComplete  1 onComplete     </code></pre>    <p>分析：</p>    <p>onNext() 会通知每个观察者，仅此而已</p>    <h3>AsyncSubject</h3>    <p><a href="/misc/goto?guid=4959738219405596915" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/e034246ad488aa7f1309ee9870f2243c.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">AsyncSubject<Integer> subscriber = AsyncSubject.create();  subscriber.subscribe(getObserver(1));  subscriber.onNext(1);     subscriber.subscribe(getObserver(2));  subscriber.onNext(2);  subscriber.onComplete();        </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">Observer1onSubscribe  Observer2onSubscribe  Observer1onNext: 2  Observer1onComplete  Observer2onNext: 2  Observer2onCompelte     </code></pre>    <p>分析：</p>    <p>查看文档，关于 AsyncObject 的介绍，在 调用 onComplete() 之前，除了 subscribe() 其它的操作都会被缓存，在调用 onComplete() 之后只有最后一个 onNext() 会生效</p>    <h3>BehaviorObject</h3>    <p><a href="/misc/goto?guid=4959738219492002090" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/11b540bfdda812ae611cac4bbcdbee6c.png"></p>    <pre>  <code class="language-java">BehaviorSubject<Integer> subscriber = BehaviorSubject.create();     subscriber.subscribe(getObserver(1));     subscriber.onNext(0);  subscriber.onNext(1);        subscriber.subscribe(getObserver(2));  subscriber.onNext(4);     subscriber.onComplete();     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">1 onSubscribe  1 onNext:0  1 onNext:1  2 onSubscribe  2 onNext:1  1 onNext:4  2 onNext:4  1 onComplete  2 onComplete     </code></pre>    <p>分析：</p>    <p>BehaviorSubject 的最后一次 onNext() 操作会被缓存，然后在 subscribe() 后立刻推给新注册的 Observer</p>    <p>对于上面的例子 onNext(1) 这个操作会被缓存，在 subscribe(observer2) 之后会立刻传入 onNext(1) 从而执行</p>    <h3>Completable</h3>    <p><a href="/misc/goto?guid=4959721339967119603" rel="nofollow,noindex">doc</a></p>    <p>先看下文档里的说明：</p>    <p>Represents a deferred computation without any value but only indication for completion or exception. The class follows a similar event pattern as Reactive-Streams: onSubscribe (onError|onComplete)?</p>    <p>也就是说 Completable 是没有 onNext 的，要么成功要么出错，不关心过程，在 subscribe 后的某个时间点返回结果</p>    <p>例子：</p>    <pre>  <code class="language-java">Completablecompletable = Completable.timer(1000, TimeUnit.MILLISECONDS);  completable.subscribe(new CompletableObserver() {      @Override      public void onSubscribe(Disposable d) {          p("onSubscribe " + System.currentTimeMillis());      }         @Override      public void onComplete() {          p("onComplete " + System.currentTimeMillis());      }         @Override      public void onError(Throwable e) {          p("onError" + e);      }  });     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">onSubscribe 1487237357951  onComplete 1487237358952     </code></pre>    <h3>Flowable</h3>    <p><a href="/misc/goto?guid=4959738219613950450" rel="nofollow,noindex">doc</a></p>    <p><img src="https://simg.open-open.com/show/e8ada8a7aea89c013f5bbb6bfcfa3770.png"></p>    <p>例子：</p>    <pre>  <code class="language-java">Flowable.just(0, 1, 2, 3)          .reduce(50, (a, b) -> a + b)          .subscribe(getSingleObserver(0));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0 onSubscribe  0 onSuccess :56     </code></pre>    <p>分析：</p>    <p>Flowable 在 RxJava2 中是用来解决背压问题的，至于具体的可以看前面推荐的文章，这里就不展开了。</p>    <h3>CompositeDisposable</h3>    <p>例子：</p>    <pre>  <code class="language-java">new CompositeDisposable().addAll(Observable.just(0, 1, 2, 3)          .subscribeOn(Schedulers.io())          .observeOn(Schedulers.computation())          .subscribeWith(getDisposableObserver(0)),     Observable.just(6, 7, 8, 9)          .subscribeOn(Schedulers.io())          .observeOn(Schedulers.computation())          .subscribeWith(getDisposableObserver(1)));     </code></pre>    <p>输出：</p>    <pre>  <code class="language-java">0  6  1  7  java.lang.InterruptedException: sleepinterrupted      atjava....  java.lang.InterruptedException: sleepinterrupted      atjava...     </code></pre>    <p>分析：</p>    <p>就是一个 Disposable 的集合</p>    <p>就这么多了，应该常用的操作符都差不多有了。</p>    <p>随便写的，如果有什么地方没说清楚欢迎留言，我会尽快回复。</p>    <p>顺便求波赞和星，谢谢。</p>    <p> </p>    <p>来自：https://blog.xujifa.cn/index.php/2017/02/20/rxjava2_introduction/</p>    <p> </p>