标签归档：RxJava

RxJava 线程模型

发表于由
回复

本文基于 RxJava 2.1.2 。根据代码和输出日志会更容易理解。

RxJava 的线程模型如下：

1. 不指定线程的情况

  • 不指定线程也就是不使用 observeOn 和 subscribeOn，所有操作在调用 subscribe 的线程执行。
@Test
public void noThread() {
    buildObservable().subscribe();
}

上面代码的输出为：

Thread[main]   execute   Action start emmit
Thread[main]   execute   Operation-1, event: 1
Thread[main]   execute   Operation-2, event: 1

2. subscribeOn

  • subscribeOn 不管调用多少次，只以第一次为准。如果只使用了 subscribeOn、没有使用 observeOn，则所有操作在第一次调用生成的线程里执行。
@Test
public void subscribeOn() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);

    Observable<Integer> observable = buildObservable();
    observable
        .subscribeOn(scheduler("subscribeOn-1"))
        .subscribeOn(scheduler("subscribeOn-2"))
        .subscribe(i -> {
            showMessageWithThreadName("Action subscribe");
            latch.countDown();
        });

    latch.await();
}

上面代码的输出为：

create scheduler subscribeOn-2
create scheduler subscribeOn-1
Thread[subscribeOn-1]   execute   Action start emmit
Thread[subscribeOn-1]   execute   Operation-1, event: 1
Thread[subscribeOn-1]   execute   Operation-2, event: 1
Thread[subscribeOn-1]   execute   Action subscribe

3. observeOn

  • observeOn 必须跟 subscribeOn 一起使用，单独使用会抛出空引用异常。
  • observeOn 应在 subscribeOn 的后面调用，否则会出现死锁的情况。
  • observeOn 操作会更改后续操作的执行线程，直至下一个 observeOn 调用之前的操作或 subscribe 操作。

继续阅读 →

RxJava

发表于由
回复

ReactiveX

ReactiveX 是一个用于异步编程的 API 规范。 ReactiveX 结合了 Observer 模式、Iterator 模式和函数式编程的最佳理念。

ReactiveX 带来了更好的代码基础：

  • Functional, 函数式：避免了复杂的有状态的程序，在可观察流上使用干净（无副作用）的 输入/输出 函数。
  • Less is more, 少即是多：ReactiveX 的操作子通常把精心制作的修改简化为几行代码。
  • Async error handling, 异步错误处理：传统的 try/catch 对于异步计算的错误非常乏力，但 ReactiveX 具有恰当的机制来处理错误。
  • Concurrency made easy, 更容易的并发：ReactiveX 的 Observables 和 Schedulers 允许程序员从底层的线程、同步和并发问题中抽象出来。

RxJava

RxJava 是 ReactiveX 在 Java 编程语言里的一个实现。

基本概念：

  • 事件：主题生成的、订阅者感兴趣的东西。
  • 订阅者：Observer，抽象基类是 Subscriber。
  • 主题：被观察的对象，抽象基类是 Observable。每个主题都有一个 OnSubscribe 的实例，OnSubscribe 从类名看是对订阅行为的反应，其 call(Subscriber subscriber) 方法封装了事件发生、通知的逻辑，供每次订阅时调用。

  • 订阅：subscribe，是一种动作，RxJava 在订阅时建立主题与监听者的关系，每次订阅，主题都会调用其内部 OnSubscribe.call(Subscriber subscriber) 方法。

  • 对于 Observable.doOnNext/doOnCompleted/doOnError/doOnEach/map 这类中间操作，生成一个新的订阅者 Subscriber，封装了相关行为，用于添加新的逻辑，并代理了对之前订阅者的调用；用新的订阅者和当前主题创建新的主题并返回。（采用的是包装器模式）
    继续阅读 →