进击Android架构师——RX响应式编程（RxLifecycl

作者: 码农的地中海 | 来源:发表于2022-07-18 18:57 被阅读0次

一、前言

RxJava简洁的编码风格、易用易读的链式方法调用、强大的异步支持等使得RxJava被广泛使用，它通过线程调度器更容易控制和切换线程，如果该工作线程还没执行结束就退出Activity或者Fragment，就会Activity或者Fragment无法释放引起内存泄漏。

比如，使用RxJava发布一个订阅后，当Activity被finish，此时订阅逻辑还未完成，如果没有及时取消订阅，就会导致Activity无法被回收，从而引发内存泄漏。

二、RxLifecycle原理分析

Android资深架构师学习【一键获取】
RxLifecycle有两种使用方法:

1、bindToLifecycle直接绑定生命周期

  Observable.interval(1, TimeUnit.SECONDS)
            .doOnDispose(new Action() {
                @Override
                public void run() throws Exception {
                    Log.i(TAG, "Unsubscribing subscription from onStart()");
                }
            })
            .compose(this.<Long>bindToLifecycle())//使用bindToLifecycle
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long num) throws Exception {
                    Log.i(TAG, "Started in onStart(), running until in onStop(): " + num);
                }
            });

2、bindUntilEvent绑定特定的ActivityEvent

Observable.interval(1, TimeUnit.SECONDS)
            .doOnDispose(new Action() {
                @Override
                public void run() throws Exception {
                    Log.i(TAG, "Unsubscribing subscription from onCreate()");
                }
            })
            .compose(this.<Long>bindUntilEvent(ActivityEvent.PAUSE))//使用bindUntilEvent
            .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long num) throws Exception {
                    Log.i(TAG, "Started in onCreate(), running until onPause(): " + num);
                }
            });

原理其实一样，只分析bindToLifecycle是怎么绑定Activity的生命周期的。
首先bindToLifecycle是通过compose操作的，compose比较常见，比如切换线程的时候，用compose操作符让代码更好看些，compose操作的是个Transformer，跟进去看,bindToLifecycle果然返回一个封装了的LifecycleTransformer：

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

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

好，这里出现了两个重要的东西lifecycleSubject和ACTIVITY_LIFECYCLE，bind方法将这两个东西处理后，返回的LifecycleTransformer。lifecycleSubject和ACTIVITY_LIFECYCLE是什么？看源码:

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

lifecycleSubject原来是个BehaviorSubject，BehaviorSubject可以当做Observable来发射消息，查看RxActivity或RxAppCompatActivity类的生命周期回调方法，lifecycleSubject都发送了该生命周期标识，比如:

protected void onPause() {
    lifecycleSubject.onNext(ActivityEvent.PAUSE);
    super.onPause();
}
protected void onStop() {
    lifecycleSubject.onNext(ActivityEvent.STOP);
    super.onStop();
}
protected void onDestroy() {
    lifecycleSubject.onNext(ActivityEvent.DESTROY);
    super.onDestroy();
}

我们来想下，一般情况下我们都是希望在Activiy退出(onDestroy)后，取消订阅消息，那么 lifecycleSubject.onNext(ActivityEvent.DESTROY)在onDestroy中发射后，就要通知该取消订阅消息了，正好takeUntil操作符可以满足要求，takeUntil操作符简要说明如下：

//Observable1发射数据，但是当Observable2也发射数据时，订阅就取消了。
Observable1
 .takeUntil(Observable2)
 .subscribe(new Consumer);

我们想要的正是在ActivityEvent.DESTROY消息发送后，订阅取消。回到上面，我们知道了lifecycleSubject是干什么用的了，lifecycleSubject就是发射生命周期标识，通过这个来监控Activity生命周期。
那ACTIVITY_LIFECYCLE是干什么用的？ACTIVITY_LIFECYCLE是个Function:

  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");
                }
            }
        };

这个Function的作用就是返回订阅消息所对应的生命周期，意思是如果你在onCreate订阅，那么就返回DESTROY，意味只有在Activity onDestroy的时候取消订阅。以下分析都依在DESTROY中取消订阅为例。
lifecycleSubject和ACTIVITY_LIFECYCLE作用明白了，继续往下走,bind方法里面是：

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

输入的lifecycleSubject和ACTIVITY_LIFECYCLE被*takeUntilCorrespondingEvent*处理了，这个方法才是RxLifecycle最精华的地方:

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);
    }

拆开看：

//这两个其实就是lifecycleSubject，可以理解为两个Observable
lifecycle.take(1).map(correspondingEvents),//map操作符调用correspondingEvents(就是上面的ACTIVITY_LIFECYCLE)的作用上面说了，类似输入CREATE，输出DESTROY
lifecycle.skip(1),//跳过之前发射的生命周期，比如在onStart里发射，跳过去，只操作onStart之后的生命周期

combineLatest操作符的作用就是操作这两个Observable（其实都是lifecycleSubject），再回到takeUntil操作符，只有在onDestroy的时候发射，才能取消订阅，而在其他生命周期，比如onPause，onStop这些不能取消，这时候我们有了第一个数据源lifecycle.take(1).map(correspondingEvents)返回的DESTROY，还有第二个数据源lifecycle.skip(1)返回的PAUSE，STOP，DESTROY这些订阅后的生命周期，那么就可以操作：只有第二个数据等于第一个数据才会取消订阅，于是回到源码,通过combineLatest的第三个参数来完成这个比较：

new BiFunction<R, R, Boolean>() {
     @Override
     public Boolean apply(R bindUntilEvent, R lifecycleEvent) throws Exception{
          return lifecycleEvent.equals(bindUntilEvent);//两个数据源比较
      }
 })

好了，到了这一步，我们得到了每个生命周期回调时，takeUntil操作符操作的Observable，类型是Boolean，只有DESTROY时返回true，其他生命周期来时返回false。但是false不能返回啊，因为takeUntil一旦把false发射出去，那订阅就取消了，我们只想在true的时候才发射出去，filter操作符上场了:

Observable\\前面已经得到的false或true,通过filter，只有true才发射
.filter(Functions.SHOULD_COMPLETE);

static final Predicate<Boolean> SHOULD_COMPLETE = new Predicate<Boolean>() {
        @Override
        public boolean test(Boolean shouldComplete) throws Exception {
            return shouldComplete;
        }
};

SHOULD_COMPLETE很简单，输入输出一样，但是完成了只有true才发射的要求。
*至此，RxLifecycle通过bindToLifecycle绑定生命周期的原理应该清楚了,再总结下:*
1、在Activiy的每个生命周期回调里通过lifecycleSubject发射当前生命周期标识,比如发射A
2、通过ACTIVITY_LIFECYCLE得到想要取消订阅的生命周期标识B
3、在takeUntilCorrespondingEvent方法里比较A和B，得到是否相等的结果C
4、通过filter操作符只发射3中结果是true的C
5、因为使用了takeUntil操作符，一旦4里发射了，那订阅就取消了。

最后takeUntil是在哪里调用的呢，回到开始的LifecycleTransformer:

public final class LifecycleTransformer<T> implements ObservableTransformer<T, T>,
                                                      FlowableTransformer<T, T>,
                                                      SingleTransformer<T, T>,
                                                      MaybeTransformer<T, T>,
                                                      CompletableTransformer
{
 @Override
    public ObservableSource<T> apply(Observable<T> upstream) {
        return upstream.takeUntil(observable);
    }
    .........
}

三、RxLifecycle解决RxJava内存泄漏问题

目前网上对RxJava的内存泄漏有几种方案：

1、通过封装，手动为RxJava的每一次订阅进行控制，在指定的时机进行取消订阅；
2、使用 Daniel Lew 的 RxLifecycle ，通过监听Activity、Fragment的生命周期，来自动断开subscription以防止内存泄漏。

笔者上述两种方式都使用过，RxLifecycle显然对于第一种方式，更简单直接，并且能够在Activity/Fragment容器的指定生命周期取消订阅，实在是好用。

方法解决：

1.）在build.gradle文件中添加依赖

// RxLifecycle基础库
compile 'com.trello.rxlifecycle2:rxlifecycle:2.1.0'
// Android组件库，里面定义了例如RxAppCompatActivity、RxFragment之类的Android组件
compile 'com.trello.rxlifecycle2:rxlifecycle-components:2.1.0'
 12341234

2.）根据不同的需要Activity继承RxActivity ，Fragment继承RxFragment

public class MainActivity7 extends RxActivity {
    private TextView mTextView;
    private static final String TAG = "RxLifecycle";
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        mTextView = (TextView) findViewById(R.id.text);
        Observable.interval(1, TimeUnit.SECONDS)
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    Log.i(TAG, "Unsubscribing subscription from onCreate()");
                }
            })
            .subscribe(new Action1<Long>() {
                @Override
                public void call(Long num) {
                    Log.i(TAG, "Started in onCreate(), running until onPause(): " + num);
                }
            });
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
    }
}
 1234567891011121314151617181920212223242526272812345678910111213141516171819202122232425262728

当activity对象被销毁，而我们的在rxjava中执行一些耗时的操作（例如：网络访问–>成功获取数据–>刷新UI；文件的存储等），就会造成内存泄露的现象。
3.）使用bindToLifecycle()的方式 —->绑定生命周期，自动取消订阅
完成Observable发布的事件和当前的组件绑定，实现生命周期同步。从而实现当前组件生命周期结束时，自动取消对Observable订阅。

    Observable.interval(1, TimeUnit.SECONDS)
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    Log.i(TAG, "Unsubscribing subscription from onCreate()");
                }
            })
            //Note:bindToLifecycle的自动取消订阅示例，因为是在onStart的时候调用，所以在onStop的时候自动取消订阅
            .compose(this.<Long>bindToLifecycle())
            .subscribe(new Action1<Long>() {
                @Override
                public void call(Long num) {
                    Log.i(TAG, "Started in onCreate(), running until onPause(): " + num);
                }
            });
    }
 1234567891011121314151612345678910111213141516

4.）使用bindUntilEvent()方式—>手动设置取消订阅的时机
使用CREATE、START、 RESUME、PAUSE、STOP、 DESTROY分别对应生命周期内的方法。使用bindUntilEvent指定在哪个生命周期方法调用时取消订阅。

             Observable.interval(1, TimeUnit.SECONDS)
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    Log.i(TAG, "Unsubscribing subscription from onCreate()");
                }
            })
            //Note:bindUntilEvent手动设置在activity onDestroy的时候取消订阅
            .compose(this.<Long>bindUntilEvent(ActivityEvent.DESTROY))
            .subscribe(new Action1<Long>() {
                @Override
                public void call(Long num) {
                    Log.i(TAG, "Started in onCreate(), running until onPause(): " + num);
                }
            });
    }

结语

rxlifecycle是trello开发的用于解决RxJava引起的内存泄漏的开源框架。

本文总结了通过RxLifeCycle解决RxJava的内存泄漏问题，同时也给我们提了一个警告，再好的框架都有它好的一面也有坏的一面，这时做好技术选型以及规避风险就很重要了。

进击Android架构师——RX响应式编程（RxLifecycl

一、前言

二、RxLifecycle原理分析

1、bindToLifecycle直接绑定生命周期

2、bindUntilEvent绑定特定的ActivityEvent

三、RxLifecycle解决RxJava内存泄漏问题

目前网上对RxJava的内存泄漏有几种方案：

方法解决：

更多Android资深架构师学习【一键获取】；本文主要讲解了其中一小节RX响应式编程的Rxlifecycle2。

结语

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android进阶之路

Android技术知识