使用RxJava中遇到的一些”坑“

前言

大家越用RxJava，越觉得它好用，所以不知不觉地发现代码里到处都是RxJava的身影。然而，RxJava也不是银弹，其中仍然有很多问题需要解决。这里，我简单地总结一下自己遇到的一些“坑”，内容上可能会比较松散。

一、考虑主线程的切换

RxJava中一个常用的使用方法是——在其他线程中做处理，然后切换到UI线程中去更新页面。其中，线程切换就是使用了observeOn()。后台下载文件，前台显示下载进度就可以使用这种方式完成。然而，实践发现这其中有坑。如果文件比较大，而下载包的粒度又比较小，这将导致很多通知积压下来，最终导致错误。

这种错误其实也是可以理解的，毕竟MainLooper是根据Message来工作的，Message过多必然会导致一些问题。当然，这还是比较想当然的想法，最终还是需要到源码中一探究竟。ObserveOn的原理在前面关于RxJava的文章已经有过分析，这里还是简单列一下代码。其中的重点还是OperatorObserveOn的内部类——ObserveOnSubscriber。其重要代码片段如下：

  static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {  final Subscriber<? super T> child;  final Scheduler.Worker recursiveScheduler;  final NotificationLite<T> on;  final boolean delayError;  final Queue<Object> queue;    final int limit;  // the status of the current stream  volatile boolean finished;  final AtomicLong requested = new AtomicLong();  final AtomicLong counter = new AtomicLong();    Throwable error;    long emitted;  // do NOT pass the Subscriber through to couple the subscription chain ... unsubscribing on the parent should  // not prevent anything downstream from consuming, which will happen if the Subscription is chained  public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child, boolean delayError, int bufferSize) {   this.child = child;   this.recursiveScheduler = scheduler.createWorker();   this.delayError = delayError;   this.on = NotificationLite.instance();   int calculatedSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;   // this fORMula calculates the 75% of the bufferSize, rounded up to the next integer   this.limit = calculatedSize - (calculatedSize >> 2);   if (UnsafeAccess.isUnsafeAvailable()) {    queue = new SpscArrayQueue<Object>(calculatedSize);   } else {    queue = new SpscAtomicArrayQueue<Object>(calculatedSize);   }   // signal that this is an async operator capable of receiving this many   request(calculatedSize);  }  void init() {   // don't want this code in the constructor because `this` can escape through the   // setProducer call   Subscriber<? super T> localChild = child;   localChild.setProducer(new Producer() {    @Override    public void request(long n) {     if (n > 0L) {      BackpressureUtils.getAndAddRequest(requested, n);      schedule();     }    }   });   localChild.add(recursiveScheduler);   localChild.add(this);  }  @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) {    RxJavaHooks.onError(e);    return;   }   error = e;   finished = true;   schedule();  }  protected void schedule() {   if (counter.getAndIncrement() == 0) {    recursiveScheduler.schedule(this);   }  } }