本文转自：http://blog.csdn.net/adzcsx2
RxJava在现在是一个非常前卫的异步框架，也是由于他非常新，所以比较难以驾驭。
像okhttp直接在onStop或者onDestroy 调用它的cancel方法就行了，但是Rxjava并没有那么简单。
因为假如每次请求都得到Observable对象，然后再onStop中unsubscribe取消，这样很不利于封装。而且会造成代码量很多，所以我找到了用rxlifecycle的解决方案。
先导包
compile ‘com.trello:rxlifecycle:0.5.0‘
compile ‘com.trello:rxlifecycle-components:0.5.0‘然后写RxActivity，原作者github上有源码，我根据自己的情况稍微修改了一下。
import android.os.Bundle;
import android.support.annotation.CallSuper;
import android.support.annotation.NonNull;
import android.support.v4.app.FragmentActivity;
import com.trello.rxlifecycle.ActivityEvent;
import com.trello.rxlifecycle.ActivityLifecycleProvider;
import com.trello.rxlifecycle.RxLifecycle;
import rx.Observable;
import rx.subjects.BehaviorSubject;
/**
* Created by a on 2016/5/6.
*/
public class RxActivity extends FragmentActivity implements ActivityLifecycleProvider {
public final BehaviorSubject<ActivityEvent> lifecycleSubject = BehaviorSubject.create();
@NonNull
@Override
public Observable<ActivityEvent> lifecycle() {
return lifecycleSubject.asObservable();
}
@NonNull
@Override
public <T> Observable.Transformer<T, T> bindUntilEvent(@NonNull ActivityEvent event) {
return RxLifecycle.bindUntilEvent(lifecycleSubject, event);
}
@NonNull
@Override
public <T> Observable.Transformer<T, T> bindToLifecycle() {
return RxLifecycle.bindActivity(lifecycleSubject);
}
@Override
@CallSuper
protected void onCreate(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();
}
}用自己的activity继承RxActivity。在网络请求的地方加上
.compose(RxLifecycle.bindUntilEvent(lifecycle(), ActivityEvent.STOP))
意思是在Stop周期的时候取消请求。
我用了上一篇博客上传图片的例子，现在是MainActivity进入SecondActivity之后，点击按钮之后上传一张图片，然后获得返回结果。
File file = new File(Environment.getExternalStorageDirectory() + "/123.png");
HttpUtil.getIImagePP().upLoadImage("c1a2b3ab56a2f218aed9b2ab3c16ce88", "be8318b73cef1c2bcafb6c8a77922436", HttpUtil.postFileParams("img", file))
//封装了在子线程中运行，主线程subscribe，下文会讲如何封装。
.compose(RxHelper.io_main(SecondActivity.this))
//.compose(RxLifecycle.bindUntilEvent(lifecycle(), ActivityEvent.STOP))
.subscribe(new RxSubscriber<Object>() {
@Override
public void _onNext(Object o) {
L.e("aaaa");
tv.setText((String) o);
}
@Override
public void _onError(String msg) {
}
});没有加.compose(RxLifecycle.bindUntilEvent(lifecycle(), ActivityEvent.STOP))这一句，点按钮上传，然后马上返回上一个activity的Log日志:
返回到MainActivity之后大概2秒钟之后，出现了这句
05-06 18:37:58.590 15642-15642/com.sanniuben.myapplication E/way: aaaa加了之后就没有返回了。
可见他在onStop的时候取消了这次请求。
RxLifecycle Github ：https://github.com/trello/RxLifecycle
RxFragment源码页
Fragment的操作也类似这样。只不过需要继承RxFragment
接下来是封装。
public class RxHelper<T> {
//子线程运行，主线程回调
//注意这里不能设置成静态，设置成静态之后泛型就不能用了。这也是我之前为什么subcribe的泛型只能为Object的原因，现在博客已经更新。
public Observable.Transformer<T, T> io_main(final RxActivity context) {
return new Observable.Transformer<T, T>() {
@Override
public Observable<T> call(Observable<T> tObservable) {
Observable<T> tObservable1 = (Observable<T>) tObservable
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Action0() {
@Override
public void call() {
//ProgressDialogUtil.showProgress中实现了弱引用，不会造成内存泄漏。
ProgressDialogUtil.showProgress(context, "正在加载，请稍候");
}
})
.subscribeOn(AndroidSchedulers.mainThread())
.observeOn(AndroidSchedulers.mainThread())
.compose(RxLifecycle.bindUntilEvent(context.lifecycle(), ActivityEvent.STOP));
return tObservable1;
}
};
}
}Subscriber
/**
* Created by a on 2016/5/6.
*/
public abstract class RxSubscriber<T> extends Subscriber<T>{
@Override
public void onCompleted() {
//完成的时候取消对话框
ProgressDialogUtil.dismiss();
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
_onError(e.getMessage());
if (!NetUtils.isConnected(MyApplication.getContextObject())) {
Toast.makeText(MyApplication.getContextObject(), "请求失败，请检查网络!", Toast.LENGTH_SHORT).show();
ProgressDialogUtil.dismiss();;
return;
}
}
@Override
public void onNext(T t) {
_onNext(t);
}
public abstract void _onNext(T t);
public abstract void _onError(String msg);
}现在来看一下，新的请求
File file = new File(Environment.getExternalStorageDirectory() + "/123.png");
HttpUtil.getIImagePP().upLoadImage("c1a2b3ab56a2f218aed9b2ab3c16ce88", "be8318b73cef1c2bcafb6c8a77922436", HttpUtil.postFileParams("img", file))
//只需要加上这一句,RxHelper<T>泛型内容需要自己输入
.compose(new RxHelper<String>.io_main(SecondActivity.this))
//这里是new的我们自己的RxSubscriber
.subscribe(new RxSubscriber<String>() {
@Override
public void _onNext(String o) {
L.e("aaaa");
tv.setText(o);
}
@Override
public void _onError(String msg) {
}
});这样就完成了rxjava的封装。现在实现了每次访问网络的时候显示对话框，请求完取消对话框，在Activity生命周期Stop的时候取消请求。
Transformer实际上就是一个Func1< Observable< T >, Observable< R>>，换言之就是：可以通过它将一种类型的Observable转换成另一种类型的Observable，和调用一系列的内联操作符是一模一样的。
compose和flatMap()的区别：
What About flatMap()?
1、At this point, you may be wondering what the difference is between using compose() and flatMap(). They both emit Observable, which means both can reuse a series of operators, right?
The difference is that compose() is a higher level abstraction: it operates on the entire stream, not individually emitted items. In more specific terms:
2、compose() is the only way to get the original Observable from the stream. Therefore, operators that affect the whole stream (like subscribeOn() and observeOn()) need to use compose().
In contrast, if you put subscribeOn()/observeOn() in flatMap(), it would only affect the Observable you create in flatMap() but not the rest of the stream.
3、compose() executes immediately when you create the Observable stream, as if you had written the operators inline. flatMap() executes when its onNext() is called, each time it is called. In other words, flatMap() transforms each item, whereas compose() transforms the whole stream.
flatMap() is necessarily less efficient because it has to create a new Observable every time onNext() is called. compose() operates on the stream as it is.
If you want to replace some operators with reusable code, use compose(). flatMap() has many uses but this is not one of them.
引用自：http://blog.danlew.net/2015/03/02/dont-break-the-chain/
简单的说就是：
1、compose是影响整条流，而flatMap仅仅是影响它自己转化的部分。
2、flatMap转后的是在调用next()后执行，而compose是直接执行。。
3、flatMap在next()后执行都会创建一个Observable 对象(以供下一次操作做处理形成链式结构)，而compose更像是拼接到流上面去，一次性执行完，所以相对来说compose的效率更高一点。
本文代码：http://download.csdn.net/detail/jdsjlzx/9529124