使用RxJava中遇到的一些”坑“_网络技术_宿州市腾雀网络科技有限公司

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

发布时间 - 2026-01-11 01:22:39 点击率：次

前言

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

一、考虑主线程的切换

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

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

 /** Observe through individual queue per observer. */
 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;
  /** The emission threshold that should trigger a replenishing request. */
  final int limit;

  // the status of the current stream
  volatile boolean finished;

  final AtomicLong requested = new AtomicLong();

  final AtomicLong counter = new AtomicLong();

  /**
   * The single exception if not null, should be written before setting finished (release) and read after
   * reading finished (acquire).
   */
  Throwable error;

  /** Remembers how many elements have been emitted before the requests run out. */
  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);
   }
  }
 }

关键点就在于这个queue成员，这个队列存放了需要进行发送给下行线程的消息。对于主线程来说，符合其实是比较重的，从消息的生产者和消费者的模式讲，过多过快的消息会导致消息阻塞。甚至，都到不了阻塞的情况，因为queue的大小会有上限，在onNext()方法中的queue.offer()可能会产生异常，这取决于queue的实现方式。但无论如何都不可能无限大，所以无法保证绝对不出异常。

解决这个问题的方法其实也很简单，可以在生产者降低消息的产生频率。也可以在消息处理的时候先不进行线程切换，而是通过判断，在必要的时候进行线程切换，比如使用runOnUIThread() 。

二、RxJava避免内存泄漏

RxJava的响应式机制本质上还是回调实现的，因此内存泄漏也是会出现的。倘若不对Subscription进行管理，内存泄漏会非常严重。对于Subscription，其实有几个比较广泛使用的方法，比如RxLifecycle，以及简单的CompositeSubscription。至于它们的使用方法，其实都非常简单，这里就不赘述了。

说到内存泄漏，就谈点题外话，动画也可能导致内存泄漏。其原因仍然是一些回调函数，这些回调函数实现的View变化的功能，但是在被撤销以后，回调函数没有取消掉，同时View可能持有Context信息，从而导致内存泄漏。最近才发现，LoadToastView这个开源库一直存在内存泄漏，其原因正如上文所说。

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作能带来一定的帮助，如果有疑问大家可以留言交流，谢谢大家对的支持。

# android # rxjava的使用 # rxjava使用 # rxjava的坑 # RxJava入门指南及其在Android开发中的使用示例 # android使用RxJava实现预加载 # android非RxJava环境下使用Handler实现预加载 # 回调 # 都是 # 会有 # 都不 # 不出 # 有很多 # 就不 # 说到 # 也很 # 有过 # 才发现 # 想当然 # 有几个 # 仍然是 # 这篇文章 # 中去 # 比较大 # 较小 # 谢谢大家 # 好用

相关栏目：【网站优化151355 】【网络推广146373 】【网络技术251813 】【 AI营销90571 】

上一篇：exec 与 exec sp_executesql 的用法及

下一篇：Quark浏览器怎样开启夜间护眼滤镜_Quark浏览器蓝光过

相关栏目网站优化
网络推广
网络技术
AI营销

最新文章 Sublime怎么一键压缩JS代码 Su sublime如何在搜索中使用正则表达式 Sublime如何设置透明窗口效果 Su mysql如何设计商品表结构_mysql css属性背景图不显示怎么办_通过检查路如何使用Golang实现排序_Golan 农历闰月是怎么回事_为合回归年加一月调整塑造《刺客信条》艾吉奥传奇的编剧离开育碧 1英里等于多少公里 1mile和km的换 css grid布局中行和列是如何定义的 PS批量旋转和翻转图片，快速校正图片方向 C# Swagger UI自定义方法 C OPPO手机九宫格和全键盘怎么切换_OP Go语言如何实现用户登录注册_Golan 1节飞行速度多少公里每小时 1节是多少公纸嫁衣8千子树第五章庙门怎么开启庙门 Laravel 多行数据编辑表单中实现逐明日之后如何提升钓鱼等级明日之后提升钓支付宝怎样查年度账单_支付宝年度账单查看 C# 多线程UI更新Dispatcher

上一篇：exec 与 exec sp_executesql 的用法及

下一篇：Quark浏览器怎样开启夜间护眼滤镜_Quark浏览器蓝光过