RxJava 的 Observable subscribe() 后会返回 Subscription（RxJava2 是在 onSubscribe(Disposable d)），我们都知道在 Activity 退出的时候要执行 subscription.unsubscribe()。每个 Observable 订阅后，都要我们写一些管理 Subscription 的代码，于是懒惰的人们发现了 RxLifecycle，只要自己的 Activity extends RxActivity（还有其他的，不一一列举），然后就只要一行代码，就可以根据 Activity 生命周期自动取消订阅

observable
    ...
    .compose(this.<>bindUntilEvent(ActivityEvent.DESTROY))
    .subscribe(...);

在 Activity 或 Fragment 里，这样写的确很方便，只是如果我们的架构是 MVP 或者 MVVM，那么在 P 层、VM 层，如何写呢？

我的设计是这样的，有一个数据提供者 DataProvider，当然是单例，比如

DataProvider.get().getNews()  --> 返回 Observable<News>
DataProvider.get().getLoginUser() --> 返回 Observable<User>

如果需要绑定当前对象

DataProvider.get().tag(this).getNews()
DataProvider.get().tag(this).getLoginUser()

这个 tag 方法的参数是 Object（你应该能想象到，这个 Object 必然是某一个 Map 的 key）
然后在 Activity onDestroy() 中

DataProvider.get().cancel(this);

既然这个 tag 方法的参数是 Object，那么 P 层、VM 层中就方便了，绑定时就直接 tag(this)，然后应该有一个 BasePresenter 或 BaseViewModel，里面写个 destroy() 方法，在 Activity onDestroy() 中调用 viewModel.destroy() 就完事了。

public class BaseViewModel {

    public void destroy() {
        DataProvider.get().cancel(this);
    }
}

所以关键是 DataProvider 到底是怎么实现的

public abstract class AbstractDataProvider implements WebService {

    protected Map<Object, AbstractDataProvider> providerMap = new WeakHashMap<>();

    private CompositeSubscription subscription = new CompositeSubscription();

    public void cancel(Object target) {
        AbstractDataProvider provider = providerMap.get(target);
        if (provider == null) {
            return;
        }
        provider.cancel();
    }

    public void cancel() {
        subscription.unsubscribe();
        for (AbstractDataProvider provider : providerMap.values()) {
            provider.cancel();
        }
    }

    private <T> Observable.Operator<T, T> bind() {
        return new Observable.Operator<T, T>() {
            @Override
            public Subscriber<? super T> call(Subscriber<? super T> subscriber) {
                subscription.add(subscriber);
                return subscriber;
            }
        };
    }

    @Override
    public Observable<News> getNewsList(String date) {
        return WebServiceProvider.get().getNewsList(date).lift(this.<News>bind());
    }

    @Override
    public Observable<TopNews> getTopNewsList() {
        return WebServiceProvider.get().getTopNewsList().lift(this.<TopNews>bind());
    }

}

DataProvider主要是比AbstractDataProvider多了tag()方法，tag()返回的是AbstractDataProvider，防止多次调用tag()

public class DataProvider extends AbstractDataProvider {

    private static class DataProviderHolder {
        private static final DataProvider INSTANCE = new DataProvider();
    }

    public static DataProvider get() {
        return DataProviderHolder.INSTANCE;
    }

    private DataProvider() {
    }

    public AbstractDataProvider tag(Object target) {
        AbstractDataProvider provider = providerMap.get(target);
        if (provider == null) {
            provider = new DataProvider();
            providerMap.put(target, provider);
        }
        return provider;
    }

}

源码，主要包含
1、MVVM
2、RxJava 生命管理
3、Retrofit 二次动态代理
4、网络请求本地缓存