RxJava 教程第一部分：入门之 关键的类

   <p> </p>    <p>Rx 有两个最基础的类型，和其他一些扩展这两种类型的类。两个核心的类为： <a href="/misc/goto?guid=4959671530546271185" rel="nofollow,noindex">Observable</a> 和 <a href="/misc/goto?guid=4959671530630520633" rel="nofollow,noindex">Observer</a> 。 <a href="/misc/goto?guid=4959671530716512119" rel="nofollow,noindex">Subject</a> 是同时继承了 Observable 和 Observer。</p>    <p>Rx 是在 <a href="http://www.open-open.com/lib/view/open1461572189562.html">Observer 模式</a> 之上建立起来的。这种模式很常见，在 Java 中有很多地方都使用了该模式，比如 JavaFx 中的 EventHandler。 这些简单的使用方式和 Rx 对比有如下区别：</p>    <ul>     <li>使用 event handler 来处理事件很难组合使用</li>     <li>无法延时处理查询事件</li>     <li>可能会导致内存泄露</li>     <li>没有标准的标示完成的方式</li>     <li>需要手工的来处理并行和多线程</li>    </ul>    <p>Observable</p>    <p>Observable 是第一个核心类。该类包含了 Rx 中的很多实现，以及所有核心的操作函数（operator、或者说 操作符）。在本系列教程中会逐步介绍每个操作函数。现在我们只需要理解 subscribe 函数即可，下面是该函数的一种定义：</p>    <pre>  <code class="language-java">public final Subscriptionsubscribe(Subscriber<? super T> subscriber)     </code></pre>    <p>该函数是用来接收 observable 发射的事件的。当事件被发射后，他们就丢给了 subscriber， subscriber 是用来处理事件的实现。这里的 Subscriber 参数实现了 Observer 接口。</p>    <p>一个 Observable 发射 三种类型的事件：</p>    <p>– Values （数据）</p>    <p>– 完成状态，告诉 Subscriber 事件（数据） 发射完毕，没有其他数据了</p>    <p>– Error， 错误状态，如果在发射数据的过程中出现错误了。会发送该事件。</p>    <p>Observer</p>    <p>Subscriber 是 Observer 的一个实现。 Subscriber 实现了其他一些额外的功能，可以作为我们实现 Observer 的基类。现在先看看 Observer 的接口定义：</p>    <pre>  <code class="language-java">interface Observer<T> {      void onCompleted();      void onError(java.lang.Throwable e);      void onNext(T t);  }     </code></pre>    <p>每次 Observable 发射事件的时候就会执行这三个对应的函数。Observer 的 onNext 函数会被调用0次或者多次，然后会调用 onCompleted 或者 onError。在 onCompleted 或者 onError 发生以后就不会再有其他事件发射出来了。</p>    <p>在使用 Rx 开发的过程中，你会看到很多 Observable，但是 Observer 出场的时候很少。但是理解 Observer 的概念是非常重要的，虽然有很多简写方式来帮助更加简洁的使用 Observer。</p>    <p>实现 Observable 和 Observer</p>    <p>你可以手工的实现 Observer 或者扩展 Observable。 在真实场景中并不需要这样做，Rx 已经提供了很多可以直接使用的工厂方法了。使用 Rx 提供的工具来创建 Observable 和 Observer 比手工实现要更加安全和简洁。</p>    <p>要订阅到一个 Observable，并不需要提供一个 Observer 示例。subscribe 函数有各种重载方法可以使用，你可以只订阅 onNext 事件，有可以只订阅 onError 事件，这样就不用提供 Observer 对象就可以接受事件了。每次只需要提供你关心的函数即可，例如 如果你不关心 error 和完成事件，则只提供 onNext 来接收每次发送的数据即可。</p>    <p>配合 <a href="http://www.open-open.com/lib/view/open1461572602015.html">Java 8 的 Lambda 表达式</a> 则使用起来代码看起来会更加简洁，所以本系列示例代码会使用 lambda 表达式，如果你不了解的话，可以先看看 <a href="http://www.open-open.com/lib/view/open1461572602015.html">掌握 Java 8 Lambda 表达式</a> 。</p>    <p>Subject</p>    <p>Subject 是 Observable 的一个扩展，同时还实现了 Observer 接口。第一眼看上去可能有点奇怪，但是在有些场合下使用 Subject 将会非常便捷。他们可以像 Observer 一样接收事件，同时还可以像 Observable 一样把接收到的事件再发射出去。这种特性非常适合 Rx 中的接入点，当你的事件来至于 Rx 框架之外的代码的时候，你可以把这些数据先放到 Subject 中，然后再把 Subject转换为一个 Observable，就可以在 Rx 中使用它们了。你可以把 Subject 当做 Rx 中的 事件管道。</p>    <p>Subject 有两个参数类型：输入参数和输出参数。这样设计是为了抽象而不是应为使用 Subject 是为了转换数据类型。转换数据应该使用转换操作函数来完成，后面我们将介绍各种操作函数。</p>    <p>Subject 有各种不同的具体实现。下面将介绍一些非常重要的实现以及他们之间的区别。</p>    <p>PublishSubject</p>    <p>PublishSubject 是最直接的一个 Subject。当一个数据发射到 PublishSubject 中时，PublishSubject 将立刻把这个数据发射到订阅到该 subject 上的所有 subscriber 中。</p>    <pre>  <code class="language-java">public static void main(String[] args) {      PublishSubject<Integer> subject = PublishSubject.create();      subject.onNext(1);      subject.subscribe(System.out::println);      subject.onNext(2);      subject.onNext(3);      subject.onNext(4);  }     </code></pre>    <p>输出的结果：</p>    <pre>  <code class="language-java">2  3  4     </code></pre>    <p>上面的 System.out::println 是 <a href="http://www.open-open.com/lib/view/open1461572602015.html">Lambda 表达式中的函数引用</a> ，如果表达式代码块只有一个函数调用，则可以直接使用函数引用来简化代码</p>    <p>可以看到，数据 1 并没有打印出来，原因是当我们订阅到 subject 的时候，1 已经发射出去了。当订阅到 subject 后就开始接收 发射到 subject 中的数据了。</p>    <p>这是我们初次看到如何使用 subscribe 函数，值得详细研究下是如何用的。 这里我们使用了一个重载的参数只有一个 <a href="/misc/goto?guid=4959671530830894945" rel="nofollow,noindex">Function</a> 类型。这个参数 Function 接收一个参数 Integer 并且没有返回值。 没有返回值的 Function 在 Rx 中被称之为 action。 可以使用下面几种方式来提供这个 Function：</p>    <ul>     <li>提供一个 Action1 的实现对象</li>     <li>使用Lambda 表达式 实现</li>     <li>使用符合该接口定义类型的 <a href="http://www.open-open.com/lib/view/open1461572602015.html">Lambda 表达式函数引用</a> 。这里 System.out::println 函数可以接受一个 Object 对象，符合 Action 的定义（接受一个参数并没有返回值），所以我们可以把该函数作为函数应用使用。 subscribe 将会使用他收到的值作为 println 函数的参数来调用 println 函数。</li>    </ul>    <p>ReplaySubject</p>    <p>ReplaySubject 可以缓存所有发射给他的数据。当一个新的订阅者订阅的时候，缓存的所有数据都会发射给这个订阅者。 由于使用了缓存，所以每个订阅者都会收到所以的数据：</p>    <pre>  <code class="language-java">ReplaySubject<Integer> s = ReplaySubject.create();    s.subscribe(v -> System.out.println("Early:" + v));  s.onNext(0);  s.onNext(1);  s.subscribe(v -> System.out.println("Late: " + v));   s.onNext(2);     </code></pre>    <p>输出结果：</p>    <pre>  <code class="language-java">Early:0  Early:1  Late: 0  Late: 1  Early:2  Late: 2     </code></pre>    <p>不管是何时订阅的，每个订阅者都收到了所有的数据。注意后一个订阅者在处理 2 之前就先收到了之前发射的数据 0和1.</p>    <p>缓存所有的数据并不是一个十分理想的情况，如果 Observable 事件流运行很长时间，则缓存所有的数据会消耗很多内存。可以限制缓存数据的数量和时间。 ReplaySubject.createWithSize 限制缓存多少个数据；而 ReplaySubject.createWithTime 限制一个数据可以在缓存中保留多长时间。</p>    <pre>  <code class="language-java">ReplaySubject<Integer> s = ReplaySubject.createWithSize(2);   s.onNext(0);  s.onNext(1);  s.onNext(2);  s.subscribe(v -> System.out.println("Late: " + v));   s.onNext(3);     </code></pre>    <p>结果：</p>    <pre>  <code class="language-java">Late: 1  Late: 2  Late: 3     </code></pre>    <p>由于指定只缓存两个数据，所以当订阅的时候第一个数据 0 就收不到了。 限制缓存的时间也是一样的情况：</p>    <pre>  <code class="language-java">ReplaySubject<Integer> s = ReplaySubject.createWithTime(150, TimeUnit.MILLISECONDS,                                                          Schedulers.immediate());  s.onNext(0);  Thread.sleep(100);  s.onNext(1);  Thread.sleep(100);  s.onNext(2);  s.subscribe(v -> System.out.println("Late: " + v));   s.onNext(3);     </code></pre>    <p>结果：</p>    <pre>  <code class="language-java">Late: 1  Late: 2  Late: 3     </code></pre>    <p>使用时间缓存创建 ReplaySubject 需要指定一个 Scheduler， Scheduler 是 Rx 中保持时间的方式。现在可以假装他不存在，不用关心他。</p>    <p>ReplaySubject.createWithTimeAndSize 则可以同时限制时间和个数。</p>    <p>BehaviorSubject</p>    <p>BehaviorSubject 只保留最后一个值。 等同于限制 ReplaySubject 的个数为 1 的情况。在创建的时候可以指定一个初始值，这样可以确保党订阅者订阅的时候可以立刻收到一个值。</p>    <pre>  <code class="language-java">BehaviorSubject<Integer> s = BehaviorSubject.create();  s.onNext(0);  s.onNext(1);  s.onNext(2);  s.subscribe(v -> System.out.println("Late: " + v));   s.onNext(3);     </code></pre>    <p>结果：</p>    <pre>  <code class="language-java">Late: 2  Late: 3     </code></pre>    <p>下面的示例只是打印出 Completed， 由于最后一个事件就是 Completed。</p>    <pre>  <code class="language-java">BehaviorSubject<Integer> s = BehaviorSubject.create();  s.onNext(0);  s.onNext(1);  s.onNext(2);  s.onCompleted();  s.subscribe(      v -> System.out.println("Late: " + v),      e -> System.out.println("Error"),      () -> System.out.println("Completed")  );     </code></pre>    <p>这里使用了 subscribe 函数的另外一种重载形式，接受三个参数。</p>    <p>下面使用了默认初始化值，如果订阅者的发射数据之前就订阅了，则会收到这个初始化的值：</p>    <pre>  <code class="language-java">BehaviorSubject<Integer> s = BehaviorSubject.create(0);  s.subscribe(v -> System.out.println(v));  s.onNext(1);     </code></pre>    <p>结果：</p>    <pre>  <code class="language-java">0  1     </code></pre>    <p>由于 BehaviorSubject 的定义就是总是有可用的数据，所以一般都会使用初始化值来创建 BehaviorSubject 。</p>    <p>AsyncSubject</p>    <p>AsyncSubject 也缓存最后一个数据。区别是 AsyncSubject 只有当数据发送完成时（onCompleted 调用的时候）才发射这个缓存的最后一个数据。可以使用 AsyncSubject 发射一个数据并立刻结束。</p>    <pre>  <code class="language-java">AsyncSubject<Integer> s = AsyncSubject.create();  s.subscribe(v -> System.out.println(v));  s.onNext(0);  s.onNext(1);  s.onNext(2);  s.onCompleted();     </code></pre>    <p>结果：</p>    <table>     <tbody>      <tr>       <td> <p>2</p> </td>      </tr>     </tbody>    </table>    <p>如果上面的示例不调用 s.onCompleted(); 则什么结果都不会打印出来。</p>    <p>隐含的规则</p>    <p>Rx 中有一些隐含的规则在代码中并不太容易看到。一个重要的规则就是当一个事件流结束（onError 或者 onCompleted 都会导致事件流结束）后就不会发射任何数据了。这些 Subject 的实现都遵守这个规则，subscribe 函数也拒绝违反该规则的情况。</p>    <pre>  <code class="language-java">Subject<Integer, Integer> s = ReplaySubject.create();  s.subscribe(v -> System.out.println(v));  s.onNext(0);  s.onCompleted();  s.onNext(1);  s.onNext(2);     </code></pre>    <p>结果：</p>    <table>     <tbody>      <tr>       <td> <p>0</p> </td>      </tr>     </tbody>    </table>    <p>但是在 Rx 实现中并没有完全确保这个规则，所以你在使用 Rx 的过程中要注意遵守该规则，否则会出现意料不到的情况。</p>    <p>来自： <a href="/misc/goto?guid=4959671530919433728" rel="nofollow">http://blog.chengyunfeng.com/?p=948</a></p>