好了，我们都知道了rxjava中的结构模式，具体应用起来是什么样子的呢？
在observable和observer之间，具体的订阅关系是怎么确立的！有以下几种方式 1.完整回调 Observer<String> observer = new Observer<String>() { @Override public void onNext(String s) { Log.d(tag, "Item: " + s); } @Override public void onCompleted() { Log.d(tag, "Completed!"); } @Override public void onError(Throwable e) { Log.d(tag, "Error!"); } };或者是 Subscriber<String> subscriber = new Subscriber<String>() { @Override public void onNext(String s) { Log.d(tag, "Item: " + s); } @Override public void onCompleted() { Log.d(tag, "Completed!"); } @Override public void onError(Throwable e) { Log.d(tag, "Error!"); } };然后完成订阅关系->observable.subscribe(observer); // 或者： observable.subscribe(subscriber);2.不完整的回调 Action1<String> onNextAction = new Action1<String>() { // onNext() @Override public void call(String s) { Log.d(tag, s); } }; Action1<Throwable> onErrorAction = new Action1<Throwable>() { // onError() @Override public void call(Throwable throwable) { // Error handling } }; Action0 onCompletedAction = new Action0() { // onCompleted() @Override public void call() { Log.d(tag, "completed"); } }; // 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext() observable.subscribe(onNextAction); // 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError() observable.subscribe(onNextAction, onErrorAction); // 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted() observable.subscribe(onNextAction, onErrorAction, onCompletedAction); 好了，关于订阅关系的确立基本就是这两种方式，下面讲一讲CompositeSubscription这个“东西” 先来看一下CompositeSubscription与Subscription究竟有什么区别 Subscription->RxJava中有个叫做Subscription的接口,可以用来取消订阅. 具体在哪取消的订阅呢？看一看源码-> public void onCompleted() { if (!done) { done = true; try { actual.onCompleted(); } catch (Throwable e) { // we handle here instead of another method so we don't add stacks to the frame // which can prevent it from being able to handle StackOverflow Exceptions.throwIfFatal(e); // handle errors if the onCompleted implementation fails, not just if the Observable fails _onError(e); } finally { // auto-unsubscribe unsubscribe(); } } }没错，就是在finally代码块里，执行的unsubscribe（）;取消订阅。说了这么半天，那CompositeSubscription呢？别着急，让我们先想一个问题，如何处理Activity的生命周期？主要就是两个问题： 1.在configuration改变（比如转屏）之后继续之前的Subscription。
比如你使用Retrofit发出了一个REST请求，接着想在listview中展示结果。如果在网络请求的时候用户旋转了屏幕怎么办？你当然想继续刚才的请求，但是怎么搞？
2.Observable持有Context导致的内存泄露
这个问题是因为创建subscription的时候，以某种方式持有了context的引用，尤其是当你和view交互的时候，这太容易发生！如果Observable没有及时结束，内存占用就会越来越大。 不幸的是，没有银弹来解决这两个问题，但是这里有一些指导方案你可以参考。
第一个问题的解决方案就是使用RxJava内置的缓存机制，这样你就可以对同一个Observable对象执行 unsubscribe/resubscribe，却不用重复运行得到Observable的代码。cache() (或者 replay())会继续执行网络请求（甚至你调用了unsubscribe也不会停止）。这就是说你可以在Activity重新创建的时候从 cache()的返回值中创建一个新的Observable对象。
Observable<Photo> request = service.getUserPhoto(id).cache(); Subscription sub = request.subscribe(photo -> handleUserPhoto(photo));注意，两次sub是使用的同一个缓存的请求。当然在哪里去存储请求的结果还是要你自己来做，和所有其他的生命周期相关的解决方案一延虎，必须在生命周期外的某个地方存储。（retained fragment或者单例等等）。
第二个问题的解决方案就是在生命周期的某个时刻取消订阅。一个很常见的模式就是使用CompositeSubscription来持有所有的Subscriptions，然后在onDestroy()或者onDestroyView()里取消所有的订阅。private CompositeSubscription mCompositeSubscription = new CompositeSubscription(); private void doSomething() { mCompositeSubscription.add( AndroidObservable.bindActivity(this, Observable.just("Hello, World!")) .subscribe(s -> System.out.println(s))); } @Override protected void onDestroy() { super.onDestroy(); mCompositeSubscription.unsubscribe(); }你可以在Activity/Fragment的基类里创建一个CompositeSubscription对象，在子类中使用它。
注意! 一旦你调用了 CompositeSubscription.unsubscribe()，这个CompositeSubscription对象就不可用了, 如果你还想使用CompositeSubscription，就必须在创建一个新的对象了。
这下我们应该能明白了，CompositeSubscription其实是Subscription的一个容载控制器。