RxLifecycle 原理解析

介绍

从名字中可以看出 RxLifecycle 和生命周期有关。也的确如此， RxLifecycle 通过监听生命周期的变化，来解决 RxJava 内存泄漏的问题。

添加依赖

1
2
3
4
5
6
7


implementation 'com.trello.rxlifecycle2:rxlifecycle:2.2.2'

// If you want to bind to Android-specific lifecycles
implementation 'com.trello.rxlifecycle2:rxlifecycle-android:2.2.2'

// If you want pre-written Activities and Fragments you can subclass as providers
implementation 'com.trello.rxlifecycle2:rxlifecycle-components:2.2.2'

基本使用

1
2
3
4
5
6
7
8
9


public class MyActivity extends RxAppCompatActivity{
  @Override
  public void onResume() {
    super.onResume();
    myObservable
        .compose(bindUntilEvent(ActivityEvent.DESTROY))
        .subscribe();
  }
}

或者

1
2
3
4
5
6
7
8
9


public class MyActivity extends RxAppCompatActivity{
  @Override
  public void onResume() {
    super.onResume();
    myObservable
        .compose(bindToLifecycle())
        .subscribe();
  }
}

使用很简单，只要继承 RxAppCompatActivity 然后调用 Observable.compose(bindToLifecycle()) 或者 Observable.compose(bindUntilEvent(ActivityEvent.DESTROY)) 就行了。

下面就从源码角度来看看 RxLifecycle 。

bindToLifecycle

就从 bindToLifecycle 看起吧

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67


public abstract class RxAppCompatActivity extends AppCompatActivity implements LifecycleProvider<ActivityEvent> {

    private final BehaviorSubject<ActivityEvent> lifecycleSubject = BehaviorSubject.create();

    @Override
    @NonNull
    @CheckResult
    public final Observable<ActivityEvent> lifecycle() {
        return lifecycleSubject.hide();
    }

    @Override
    @NonNull
    @CheckResult
    public final <T> LifecycleTransformer<T> bindUntilEvent(@NonNull ActivityEvent event) {
        return RxLifecycle.bindUntilEvent(lifecycleSubject, event);
    }

    @Override
    @NonNull
    @CheckResult
    public final <T> LifecycleTransformer<T> bindToLifecycle() {
        return RxLifecycleAndroid.bindActivity(lifecycleSubject);
    }

    @Override
    @CallSuper
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        lifecycleSubject.onNext(ActivityEvent.CREATE);
    }

    @Override
    @CallSuper
    protected void onStart() {
        super.onStart();
        lifecycleSubject.onNext(ActivityEvent.START);
    }

    @Override
    @CallSuper
    protected void onResume() {
        super.onResume();
        lifecycleSubject.onNext(ActivityEvent.RESUME);
    }

    @Override
    @CallSuper
    protected void onPause() {
        lifecycleSubject.onNext(ActivityEvent.PAUSE);
        super.onPause();
    }

    @Override
    @CallSuper
    protected void onStop() {
        lifecycleSubject.onNext(ActivityEvent.STOP);
        super.onStop();
    }

    @Override
    @CallSuper
    protected void onDestroy() {
        lifecycleSubject.onNext(ActivityEvent.DESTROY);
        super.onDestroy();
    }
}

在 bindToLifecycle 中只是调用了 RxLifecycleAndroid.bindActivity(lifecycleSubject); ，把 lifecycleSubject 当做参数传递进去，最后返回一个 LifecycleTransformer 。

这个 lifecycleSubject 继承了 Subject ， Subject 继承了 Observable 并实现了 Observer ，也就是 lifecycleSubject 拥有发射事件的能力和接收事件的能力。 lifecycleSubject 会在 Activity 的生命周期发射对应的事件，这里大胆的猜测一下 RxLifecycle 会在 Activity 销毁的时候进行取消订阅的操作。

接着来看 RxLifecycleAndroid.bindActivity 里做了什么

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30


// 1.
public static <T> LifecycleTransformer<T> bindActivity(@NonNull final Observable<ActivityEvent> lifecycle) {
  return bind(lifecycle, ACTIVITY_LIFECYCLE);
}

// 2.
public static <T, R> LifecycleTransformer<T> bind(@Nonnull Observable<R> lifecycle,@Nonnull final Function<R, R> correspondingEvents) {
        checkNotNull(lifecycle, "lifecycle == null");
        checkNotNull(correspondingEvents, "correspondingEvents == null");
        return bind(takeUntilCorrespondingEvent(lifecycle.share(), correspondingEvents));
}

// 3.
public static <T, R> LifecycleTransformer<T> bind(@Nonnull final Observable<R> lifecycle) {
  return new LifecycleTransformer<>(lifecycle);
}

private static <R> Observable<Boolean> takeUntilCorrespondingEvent(final Observable<R> lifecycle,final Function<R, R> correspondingEvents) {
        return Observable.combineLatest(
            lifecycle.take(1).map(correspondingEvents),
            lifecycle.skip(1),
            new BiFunction<R, R, Boolean>() {
                @Override
                public Boolean apply(R bindUntilEvent, R lifecycleEvent) throws Exception {
                    return lifecycleEvent.equals(bindUntilEvent);
                }
            })
            .onErrorReturn(Functions.RESUME_FUNCTION)
            .filter(Functions.SHOULD_COMPLETE);
}

在 2 中 bind 里调用了 takeUntilCorrespondingEvent ，它会根据 lifecycle（其实就是 lifelifecycleSubject） 发射的第一个值，也就是生命周期里的事件，从 correspondingEvents 取出对应要取消订阅的生命周期事件。这里的 correspondingEvents 就是 ACTIVITY_LIFECYCLE ，我们稍后再讲 ACTIVITY_LIFECYCLE 。接着 lifecycle.skip(1) 跳过了第一个，也就是第一个发射事件不能作为取消事件，如果第一个就取消了也没有什么意义。然后判断接下来的事件是不是和要取消订阅事件相等，相等才会发射事件，如果不是就不发射事件。这就起到一个过滤的作用，只会在特定的事件里去取消订阅。

最后 3 里调用了 bind(@Nonnull final Observable<R> lifecycle) 构造了 LifecycleTransformer 。我们稍后看 LifecycleTransformer 先来从 ACTIVITY_LIFECYCLE 看看发射事件和取消订阅事件有什么规律。

ACTIVITY_LIFECYCLE

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


private static final Function<ActivityEvent, ActivityEvent> ACTIVITY_LIFECYCLE =
      new Function<ActivityEvent, ActivityEvent>() {
              @Override
              public ActivityEvent apply(ActivityEvent lastEvent) throws Exception {
                  switch (lastEvent) {
                      case CREATE:
                          return ActivityEvent.DESTROY;
                      case START:
                          return ActivityEvent.STOP;
                      case RESUME:
                          return ActivityEvent.PAUSE;
                      case PAUSE:
                          return ActivityEvent.STOP;
                      case STOP:
                          return ActivityEvent.DESTROY;
                      case DESTROY:
                          throw new OutsideLifecycleException("Cannot bind to Activity lifecycle when outside of it.");
                      default:
                          throw new UnsupportedOperationException("Binding to " + lastEvent + " not yet implemented");
                  }
              }
      };

根据上面可以发现如下规律

发射的事件	对应的取消订阅的事件
CREATE	DESTORY
START	STOP
RESUME	PAUSE
PAUSE	STOP
STOP	DESTORY

在 ACTIVITY_LIFECYCLE 中发现并不是所有的事件都是在 DESTORY 进行取消订阅的。

LifecycleTransformer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


public final class LifecycleTransformer<T> implements ObservableTransformer<T, T>,
                                           FlowableTransformer<T, T>,
                                           SingleTransformer<T, T>,
                                           MaybeTransformer<T, T>,
                                           CompletableTransformer
{
  final Observable<?> observable;

  LifecycleTransformer(Observable<?> observable) {
    checkNotNull(observable, "observable == null");
    this.observable = observable;
  }

  @Override
  public ObservableSource<T> apply(Observable<T> upstream) {
    return upstream.takeUntil(observable);
  }
  ……
}

之前提到在 bindToLifecycle 最后的返回值是 LifecycleTransformer ， LifecycleTransformer 实现了 ObservableTransformer 、 FlowableTransformer 、 SingleTransformer 、 MaybeTransformer 、 CompletableTransformer ，这些大同小异，我们就看最熟悉的 ObservableTransformer 中的 apply 方法。

apply 方法很简单就只调用了 upstream.takeUntil(observable) ， takeUntil 这个操作符会在 observable 发射了一项数据或者终止事件时，取消 upstream 的订阅，丢弃所有事件。 upstream 就是我们进行网络请求的 Observable 。

所以到这里就可以很明白的知道 RxLifecycle 是在满足某个条件（这里就是生命周期中的某个环节）下，通过取消订阅来保证 RxJava 内存不会泄漏的。

bindUntilEvent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31


public static <T, R> LifecycleTransformer<T> bindUntilEvent(@Nonnull final Observable<R> lifecycle,
                                                            @Nonnull final R event) {
    checkNotNull(lifecycle, "lifecycle == null");
    checkNotNull(event, "event == null");
    return bind(takeUntilEvent(lifecycle, event));
}

private static <R> Observable<R> takeUntilEvent(final Observable<R> lifecycle, final R event) {
    return lifecycle.filter(new Predicate<R>() {
        @Override
        public boolean test(R lifecycleEvent) throws Exception {
            return lifecycleEvent.equals(event);
        }
    });
}

/**
 * Binds the given source to a lifecycle.
 * <p>
 * This helper automatically determines (based on the lifecycle sequence itself) when the source
 * should stop emitting items. Note that for this method, it assumes <em>any</em> event
 * emitted by the given lifecycle indicates that the lifecycle is over.
 *
 * @param lifecycle the lifecycle sequence
 * @return a reusable {@link LifecycleTransformer} that unsubscribes the source whenever the lifecycle emits
 */
@Nonnull
@CheckReturnValue
public static <T, R> LifecycleTransformer<T> bind(@Nonnull final Observable<R> lifecycle) {
    return new LifecycleTransformer<>(lifecycle);
}

经过上面的分析，这里就简单了。 bindUntilEvent 和 bindToLifecycle 不同的地方在于 bindUntilEvent 指定了要取消订阅的事件。这样就不用去查找第一个发射的事件对应的取消订阅的事件，而是直接把发射事件和指定的事件进行比对，如果是就取消订阅。

总结

RxLifecycle 创建了 BehaviorSubject 来发射对应的生命周期事件，通过 compose 把网络请求或者其他 Observable 关联在一起。 BehaviorSubject 在收到生命周期事件的时候会和对应取消订阅的事件进行比对，如果相等，则发射一个事件，这个事件用来告诉网络请求订阅需要被取消。取消订阅事件可以通过 bindUntilEvent 来指定。

取消订阅的原理是通过 takeUntil 操作符，当 BehaviorSubject 发出一个事件或者终止事件时，取消网络请求的订阅，这样请求的数据就不会被处理，也就不会回到主线程进行相关操作了。

参考

解决 RxJava 内存泄漏（前篇）：RxLifecycle 详解及原理分析 - 简书

文章目录

介绍