RxBus Practice With RxLifecycle

(Updated: )

概述

项目地址：

https://github.com/yueban/RxBus-with-RxLifecycle

RxBus 实现参考：

https://github.com/SmartDengg/RxWeather

RxBus 实现

RxBus 内维护两个 Bus 实例，rxBusStandard 服务于普通消息，rxBusSticky 服务于粘滞消息。

1
2
private SerializedSubject<Object, Object> rxBusStandard;
private SerializedSubject<Object, Object> rxBusSticky;

RxBus 类内部暴露两个消息发送的方法，分别对应普通消息与粘滞消息。

1
2
3
4
5
6
public static void postEvent(Object event) {
    getInstance().rxBusStandard.onNext(event);
}
public static void postEventSticky(Object event) {
    getInstance().rxBusSticky.onNext(event);
}

相应的，消息的订阅方法也有两个。

1
2
3
4
5
6
public static Observable<Object> toObservable() {
    return getInstance().rxBusStandard.asObservable().onBackpressureBuffer();
}
public static Observable<Object> toObservableSticky() {
    return getInstance().rxBusSticky.asObservable().share().onBackpressureBuffer();
}

发送消息与订阅消息代码示例。

1
2
3
4
5
6
7
8
9
10
11
12
//发送消息
RxBus.postEvent(new Object());

//订阅消息
RxBus
  .toObservable()
  .subscribe(new Action1<Object>() {
      @Override
      public void call(Object o) {
          //handle event
      }
  });

如上，基于 RxJava 实现的 RxBus 代码简洁，使用方便，但存在两个缺陷。

Sticky 事件的唯一性

RxBus 内存泄漏问题

上文的示例代码中 new Actioin 持有了界面的引用，其本身又被全局单例、近乎与 进程同生命周期的 RxBus 所持有，因此会导致界面无法被回收，从而引发内存泄露。解决内存泄露的示例代码如下，在界面销毁时取消订阅即可，然而这种方案不仅要存储一个 subscription，还要手动取消订阅，额外增加了代码的维护成本，下面即将登场的 RxLifeCycle 却能很优雅的解决这个问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//订阅事件
Subscription subscription = RxBus
        .toObservable()
        .subscribe(new Action1<Object>() {
            @Override
            public void call(Object o) {
                //handle event
            }
        });

//界面被销毁时取消订阅
if (!subscription.isUnsubscribed()) {
    subscription.unsubscribe();
}

RxLifeCycle 解决内存泄露

RxLifeCycle 由 trello 开源，基于 RxJava 实现了一套自动取消订阅 Observable 的组件。官方示例中的代码很简单，调用 compose 方法传入 bindToLifecycle 得到的 Transformer，则 Observable 会在当期界面生命周期结束时自动取消订阅。

1
2
3
4
5
6
7
8
9
public class MyActivity extends RxActivity {
    @Override
    public void onResume() {
        super.onResume();
        myObservable
            .compose(bindToLifecycle())
            .subscribe();
    }
}

考虑到实际项目中我们通常会自己维护一个 BaseActivity，RxLifeCycle 当然也考虑到了这种情况，不想继承 RxActivity，只需要自己的 Activity 实现 ActivityLifecycleProvider 接口就行，这个实现也很简单，直接复制 RxActivity 中的代码即可。关于 RxLifeCycle 的原理与实现，感兴趣的同学可以自己阅读源码，并不复杂，本文不再赘述。

如果需要指定声明周期中的某个阶段取消订阅，则可以通过下面的代码实现。

1
2
3
4
//在 Activity 生命周期中的 STOP 阶段取消订阅
myObservable
    .compose(bindUntilEvent(ActivityEvent.STOP))
    .subscribe();

Demo

三个界面

  • MainActivity
  • Activity2
  • Activity3

每个界面在 onCreate() 方法中执行如下操作

  1. 发送一条 stickyEvent
  2. 发送一条普通 Event
  3. 注册 stickyEventBus
  4. 注册普通 EventBus

消息接收后会打印 Log，Tag 为当前界面，内容为 Event 附带信息。Event 附带信息包含 发送界面 与 消息类型 两部分，如 MainActivity_sticky 表示 MainActivity 发出的 stickyEvent。

每个界面中有一个 Button，点击后跳转界面，顺序为 MainActivity → Activity2 → Activity3 -(clearTop)→ MainActivity，因此我们的预期结果应为：

  1. MainActivity 接收到 MainActivity 发送的 stickyEvent
  2. MainActivity 接收到 Activity2 发送的 stickyEvent
  3. MainActivity 接收到 Activity2 发送的 normalEvent
  4. Activity2 接收到 Activity2 发送的 stickyEvent
  5. MainActivity 接收到 Activity3 发送的 stickyEvent
  6. Activity2 接收到 Activity3 发送的 stickyEvent
  7. MainActivity 接收到 Activity3 发送的 normalEvent
  8. Activity2 接收到 Activity3 发送的 normalEvent
  9. Activity3 接收到 Activity3 发送的 stickyEvent

程序运行后输出 Log 如下，与预期相符。

Flux Package Architecture

Demo 拓展

我们基于 RxLifeCycle 使 MainActivity 中的 Bus 在 onPause 后取消订阅，则输出结果应减少上述 2，3，5，7 四步，看看运行后的结果，当然也与预期相符，同时证实了 RxLifeCycle 的可靠性。

Flux Package Architecture

源码地址请看文章 开头。