RxJava 从入门到爱上它 - 线程控制的深入理解

作者: Nichool | 来源:发表于2017-05-27 15:41 被阅读0次

<a href="http://www.jianshu.com/p/808b09d1b5ab">RxJava 从入门到爱上它 - 认识</a>
<a href="http://www.jianshu.com/p/ffb6c25f472b">RxJava 从入门到爱上它 - "变换"的深入理解</a>
关于Backpressure 作者也将在后续的文章中做出解析。

RxJava 的线程控制使用与原理

RxJava中的线程控制有两个方法

subscribeOn : 发生在通知流程(如果对通知流程与发送流程不理解的请查看我的《"变换"深入理解》)，指定的是当前Observable.call()方法以及之后所有操作的线程。(下面将会详细分析这一方法的作用)
observeOn : 发生在发送流程，指定的是当前Subscriber.onNext()方法以及之后的所有操作的线程。(同上)

此方法有一个单参数：

调度器类型	效果
Schedulers.computation()	用于计算任务，如事件循环或和回调处理，不要用于IO操作(IO操作请使用Schedulers.io())；默认线程数等于处理器的数量
Schedulers.from(executor)	使用指定的Executor作为调度器
Schedulers.immediate()	在当前线程立即开始执行任务
Schedulers.io()	用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长；对于普通的计算任务，请使用Schedulers.computation()；Schedulers.io()默认是一个CachedThreadScheduler，很像一个有线程缓存的新线程调度器
Schedulers.newThread()	为每个任务创建一个新线程
Schedulers.trampoline()	当其它排队的任务完成后，在当前线程排队开始执行

一个常见的线程控制的例子：

Observable.just("hello"， "my"， "name"， "is"， "nichool")
               .map(new Func1<String， String>() {
                   @Override
                   public String call(String s) {
                      //在io线程中 执行
                       LogUtils.LogW("return " + " map: " + s + " " + Thread.currentThread().getName());
                       return " map: " + s;
                   }
               })
               .subscribeOn(Schedulers.io())
               .observeOn(AndroidSchedulers.mainThread())
               .subscribe(new Subscriber<String>() {
                   @Override
                   public void onStart() {
                       super.onStart();
                       //在主线程中执行
                       LogUtils.LogW(" onStart " + Thread.currentThread().getName());
                   }

                   @Override
                   public void onCompleted() {
                       //在主线程中执行
                       LogUtils.LogW(" onCompleted " + Thread.currentThread().getName());
                   }

                   @Override
                   public void onError(Throwable e) {
                       LogUtils.LogW(" onError " + Thread.currentThread().getName());
                   }

                   @Override
                   public void onNext(String s) {
                       //在主线程中执行
                       LogUtils.LogW(" onNext( " + s + " ) " + Thread.currentThread().getName());
                   }
               });

打印的Log:

RxJavaDemo:  onStart main
RxJavaDemo: return  map: hello RxIoScheduler-2
RxJavaDemo: return  map: my RxIoScheduler-2
RxJavaDemo:  onNext(  map: hello ) main
RxJavaDemo: return  map: name RxIoScheduler-2
RxJavaDemo:  onNext(  map: my ) main
RxJavaDemo: return  map: is RxIoScheduler-2
RxJavaDemo:  onNext(  map: name ) main
RxJavaDemo: return  map: nichool RxIoScheduler-2
RxJavaDemo:  onNext(  map: is ) main
RxJavaDemo:  onNext(  map: nichool ) main
RxJavaDemo:  onCompleted main

为了更好在使用以上两种方法来控制线程，我们就来分析下原理吧。

在RxJava中所有的操作都是一次"变换" (关于"变换"的理解请查看我的《"变换"深入理解》)，同样subscribeOn与observeOn也是一种"变换"。

废话不多说了，快请上我源码大人！！

subscribeOn 源码分析：

//Observable.java
public final Observable<T> subscribeOn(Scheduler scheduler) {
     if (this instanceof ScalarSynchronousObservable) {
         return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
     }
     return create(new OperatorSubscribeOn<T>(this， scheduler));
 }

 //OperatorSubscribeOn.java
 @Override
 public void call(final Subscriber<? super T> subscriber) {
     final Worker inner = scheduler.createWorker();
     subscriber.add(inner);

     inner.schedule(new Action0() {
         @Override
         public void call() {
             final Thread t = Thread.currentThread();

             Subscriber<T> s = new Subscriber<T>(subscriber) {
                 @Override
                 public void onNext(T t) {
                     subscriber.onNext(t);
                 }

                 @Override
                 public void onError(Throwable e) {
                     try {
                         subscriber.onError(e);
                     } finally {
                         inner.unsubscribe();
                     }
                 }

                 @Override
                 public void onCompleted() {
                     try {
                         subscriber.onCompleted();
                     } finally {
                         inner.unsubscribe();
                     }
                 }

                 ...
             };

             source.unsafeSubscribe(s);
         }
     });
 }

 //Observable.java
 public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
      try {
          // new Subscriber so onStart it
          subscriber.onStart();
          // allow the hook to intercept and/or decorate
          hook.onSubscribeStart(this， onSubscribe).call(subscriber);
          return hook.onSubscribeReturn(subscriber);
      } catch (Throwable e) {
          ...
      }
  }

通过上面的源码我们得知流程如下：
subscribeOn 通过Create(new OperatorSubscribeOn<T>(this， scheduler))方法创建一个新的Observable，这个新的Observable的call方法中，通过指定的Scheduler开启任务，任务中创建一个新的Subscriber，然后通知父Observable来处理Subscriber。

总结：
subscribeOn方法创建了一个新的Observable（被观察者），当这个新的Observable被通知的时候，在指定线程中执行后面的操作（封装subscriber，通知父Observable）。其实分析到这里也就解释了subscribeOn就是一种"变换"，只不过具体操作不一样

上面源码中有这么一段逻辑，作者我查看ScalarSynchronousObservable源码得知只有在Observable.just(T t).subscribeOn()时会进入，这段逻辑内部也包含上面类似的逻辑，只不过加入了针对的处理。

   if (this instanceof ScalarSynchronousObservable) {
       return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
   }

observeOn 源码分析：

//Observable.java
public final Observable<T> observeOn(Scheduler scheduler， boolean delayError， int bufferSize) {
      if (this instanceof ScalarSynchronousObservable) {
          return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
      }
      return lift(new OperatorObserveOn<T>(scheduler， delayError， bufferSize));
  }

//OperatorObserveOn.java
@Override
public Subscriber<? super T> call(Subscriber<? super T> child) {
    if (scheduler instanceof ImmediateScheduler) {
        // avoid overhead， execute directly
        return child;
    } else if (scheduler instanceof TrampolineScheduler) {
        // avoid overhead， execute directly
        return child;
    } else {
        ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler， child， delayError， bufferSize);
        parent.init();
        return parent;
    }
}

//OperatorObserveOn$ObserveOnSubscriber.java
@Override
public void onNext(final T t) {
    if (isUnsubscribed() || finished) {
        return;
    }
    if (!queue.offer(on.next(t))) {
        onError(new MissingBackpressureException());
        return;
    }
    schedule();
}

@Override
public void onCompleted() {
    if (isUnsubscribed() || finished) {
        return;
    }
    finished = true;
    schedule();
}

@Override
public void onError(final Throwable e) {
    if (isUnsubscribed() || finished) {
        RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
        return;
    }
    error = e;
    finished = true;
    schedule();
}

protected void schedule() {
    if (counter.getAndIncrement() == 0) {
        recursiveScheduler.schedule(this);
    }
}

// only execute this from schedule()
@Override
public void call() {
    //recursiveScheduler.schedule(this) 在这里执行
    ...
    final Subscriber<? super T> localChild = this.child;
    ...
    for (;;) {
        long requestAmount = requested.get();

        while (requestAmount != currentEmission) {
            boolean done = finished;
            Object v = q.poll();
            boolean empty = v == null;

            if (empty) {
                break;
            }

            localChild.onNext(localOn.getValue(v));
            ...
        }
    }
}

通过上面的源码我们得知流程如下：
observeOn 通过lift(new OperatorObserveOn<T>()) 创建一个新的Observable(left()不再赘述, 请查看我的《"变换"深入理解》），在OperatorObserveOn.call方法中当判断scheduler为当前线程则不予处理直接返回，其他情况则通过ObserveOnSubscriber来封装传入的Subscriber生成一个新的Subscriber。在新的Subscriber的实现中，都会调用schedule()方法来指定线程执行Subscriber.onNext() (调用this.call()).

总结 (重点)

observeOn 和 subscribeOn 一样都是指定线程来完成后面的逻辑，区别在于subscribeOn是作用在通知时Observable.call()，也就是它通知流程的后面逻辑（可能包括发送流程），observeOn是作用在发送时Subscriber.onNext()。也就是它的发送流程后面的逻辑。

综上所述可以合并成这样一张流程图 (此图片来源于 https://gank.io/post/560e15be2dca930e00da1083#toc_15 )：

线程控制图

右边是对应的代码，左边流程图是具体的内部变化。不同颜色的线代表着在不同的线程中执行，向上是通知流程Observable.call()，向下是发送流程Subscriber.next()。特别注意:请看红色的线，由于subscribeOn导致切换了线程，并且后面没有observeOn切换线程，所以第一个left()将在subscribeOn指定的线程中执行。

请大家结合上面的分析以及这个流程图好好理解下RxJava的线程控制。

延展

有这样一个常见的问题，请看下面的代码。

new Thread() {
  @Override
  public void run() {
      super.run();
      Observable.just("hello"， "my"， "name"， "is"， "nichool")
              .map(new Func1<String， String>() {
                  @Override
                  public String call(String s) {
                      //io线程中执行
                      LogUtils.LogW("return " + " map: " + s + " " + Thread.currentThread().getName());
                      return " map: " + s;
                  }
              })
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(new Subscriber<String>() {
                  @Override
                  public void onStart() {
                      //new Thread 中执行
                      super.onStart();
                      LogUtils.LogW(" onStart " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onCompleted() {
                      //主线程中执行
                      LogUtils.LogW(" onCompleted " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onError(Throwable e) {
                      //主线程中执行
                      LogUtils.LogW(" onError " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onNext(String s) {
                      //主线程中执行
                      LogUtils.LogW(" onNext( " + s + " ) " + Thread.currentThread().getName());
                  }
              });
  }
}.start();

Log日志：

05-23 06:59:55.577 30333-31007/me.nieyihe.rxjavademo W/RxJavaDemo: onStart Thread-8
05-23 06:59:55.577 30333-30401/me.nieyihe.rxjavademo W/RxJavaDemo: return  map: hello RxIoScheduler-2
...
05-23 06:59:55.578 30333-30333/me.nieyihe.rxjavademo W/RxJavaDemo:  onCompleted main

可以看出上面的onStart是运行在当前线程上的(onStart是在subscribe()时就被调用的)，也就是不在主线程上的。如果我们需要在onStart中执行Android上UI的显示(比如进度条显示出来并且设置进度为0)，则直接会报错误。

解决办法: 使用doOnSubscribe与subscribeOn的组合在通知流程中额外添加一个处理来代替onStart。

new Thread() {
  @Override
  public void run() {
      super.run();
      Observable.just("hello"， "my"， "name"， "is"， "nichool")
              .subscribeOn(Schedulers.io())
              .doOnSubscribe(new Action0() {
                  @Override
                  public void call() {
                      //此处运行在主线程中 此处的call() 在Observable.call时就会调用。
                      LogUtils.LogW(" doOnSubscribe " + Thread.currentThread().getName());
                  }
              })
              .map(new Func1<String， String>() {
                  @Override
                  public String call(String s) {
                      //此处运行在IO线程中 此处的call()只有在发送流程时才会被调用。
                      LogUtils.LogW("return " + " map: " + s + " " + Thread.currentThread().getName());
                      return " map: " + s;
                  }
              })
              .subscribeOn(AndroidSchedulers.mainThread())
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(new Subscriber<String>() {
                  @Override
                  public void onStart() {
                      super.onStart();
                      LogUtils.LogW(" onStart " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onCompleted() {
                      LogUtils.LogW(" onCompleted " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onError(Throwable e) {
                      LogUtils.LogW(" onError " + Thread.currentThread().getName());
                  }

                  @Override
                  public void onNext(String s) {
                      LogUtils.LogW(" onNext( " + s + " ) " + Thread.currentThread().getName());
                  }
              });
        }
    }.start();

原理 :

doOnSubscribe创建了一个Observable，当执行通知流程时会调用这个Observable.call()方法，内部会调用这个重写的call方法。
subscribeOn（AndroidSchedulers.mainThread() 与 subscribeOn(Schedulers.io())使线程切换了两次，先是切换成AndroidSchedulers.mainThread()，然后执行完doOnSubscribe中的call()，又切换成Schedulers.io()，然后进入发送流程，然后执行map操作。

上述分析请结合更详细的源码，这样有有助于更好地理解
分析过程中如若存在错误，请在下方评论处给予指正。

↓↓↓↓↓如果觉得可以的话，请点击下面的喜欢。或者打赏哦，我会更加努力的。↓↓↓↓↓

RxJava 从入门到爱上它 - 线程控制的深入理解

RxJava 的线程控制使用与原理

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

程序员

技术干货

Android开发

@IT·互联网