本文所涉及到的RxJava操作符：

1. compose
   你完全可以将apply中的内容转移到外部，因为它并不产生新的上游数据源，通常使用它来包含一系列通用的对上游数据源的处理
2. takeUntil
   直到某一种条件达成时，会发送onComplete结束数据流。
3. combineLatest
   联合多个上游数据源，当有一个上游发送数据时，就会将所有上游所发送的最新的数据联合发送。
4. filter
   只有符合过滤条件的上游数据才会被继续发送
5. take
   只有前n个数量的上游数据才会被发送
6. skip
   从n+1个数量的上游数据开始才会被发送

直接上代码，以下代码为根据RxLifeCycle的实现提取出的一个完整的上游数据源到下游数据源所经过的历程
，注释已经写的差不多了

Ob
 .compose(new ObservableTransformer<Object, Object>() {
                            @Override
                            public ObservableSource<Object> apply(Observable<Object> upstream) {
                                //一个subject既是Observable又是Observer
                                Observable<ActivityEvent> activityEventObservable = Activity.this.provideLifecycleSubject()
                                        .share();

                                return upstream
                                        //takeUntil有两种实现
                                        // 一种是：当返回true时，就发送onComplete
                                        // 另一种是：当（参数指定的）数据源发射事件时，发送onComplete
                                        //这里采用第二种，同时借助filter操作符（控制数据源发送的事件是否会被拦截），实现当到达对应生命周期时发送onComplete的效果
                                        .takeUntil(
                                                //每当联合的数据源中，有数据源发送新的事件时，就会将所有的数据源发送的最新事件合并起来，发送一个新的事件
                                                //这里利用这个特性，在每个生命周期方法的最后调用onNext发送生命周期事件
                                                Observable.combineLatest(
                                                        //将订阅时的生命周期转化为停止发射事件的生命周期
                                                        activityEventObservable
                                                                .take(1)
                                                                .map(new Function<ActivityEvent, ActivityEvent>() {
                                                                    @Override
                                                                    public ActivityEvent apply(ActivityEvent activityEvent) throws Exception {
                                                                        return activityEvent;
                                                                    }
                                                                }),
                                                        //跳过订阅时的生命周期，当后续生命周期发送
                                                        activityEventObservable
                                                                .skip(1),
                                                        new BiFunction<ActivityEvent, ActivityEvent, Boolean>() {
                                                            @Override
                                                            public Boolean apply(ActivityEvent bindUntilEvent, ActivityEvent lifecycleEvent) throws Exception {
                                                                return lifecycleEvent.equals(bindUntilEvent);
                                                            }
                                                        })
                                                        //是否拦截事件
                                                        .filter(b -> b));
                            }
                        })