RxJava RxAndroid 入门使用用法 参照博文自己领悟使用解析

原创：转载请注明出处！！http://blog.csdn.net/wuqilianga/article/details/62423661

首先

RxJava 个人觉得太过于抽象了，不去多操练，多使用，多摸索，多巩固是学会了就忘，一看还是不懂，再看懵懂的感觉

这个是个好东西，能简化代码，加快编程提交代码的进度，也能很好地理解响应性编程的好处

首先这个东西我个人理解的方式是这样

由 两个东西  +  一个关系组成

第一个东西：发送端

第二个东西：接收端

一个关系 ： 订阅关系



其中

发送端（Observable） --  发送前执行器（doOnSubscribe） --  发送器（just、from、OnSubscribe....） 

   -- 作用环境（subscribeOn【指定发送器作用环境】；observeOn【指定下一个接收器作用环境】）

发送的途中  --
 转换发送端(lift / compose) --转换分解发送数据(map,flatMap)---转换分解集体发送端() [b] -- 
  作用环境(同发送端一样)[/b]

接收端（Subscriber）---接收器
（onNext/onError/onComplete/call.......） --    作用环境(同发送端一样)

下边是学习操作参考例子：

package com.xxx.swiperefreshlayoutdemo.rxjava;

import com.xxx.swiperefreshlayoutdemo.okhttp_gson_pager.Okhttp_Gson_Pager_Data;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observable.Operator;
import rx.Scheduler;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

/**
* Created by xxx on 2017/3/15.
*/

public class RxJavaMain {

private static  String TAG = RxJavaMain.class.getSimpleName();

private static void method1() {
//这是发送端
Observable<Okhttp_Gson_Pager_Data> mOb = Observable.create(
//发送器
new OnSubscribe<Okhttp_Gson_Pager_Data>(){
@Override
public void call(Subscriber<? super Okhttp_Gson_Pager_Data> subscriber) {
subscriber.onNext(getData());
subscriber.onCompleted();
}
});

//这是接收端
Subscriber<Okhttp_Gson_Pager_Data> mSb = new Subscriber<Okhttp_Gson_Pager_Data>() {
//接收器
@Override
public void onCompleted() {
System.out.println("hello 我是er结尾的接收端onCompleted方法！");
}

@Override
public void onError(Throwable e) {
System.out.println("hello 我是er结尾的接收端onError方法！");
}

@Override
public void onNext(Okhttp_Gson_Pager_Data obj) {
System.out.println("hello 我是er结尾的接收端onNext方法！" + " 发送端发送的数据：" + obj.getError().getErr_code());
}
};

//建立联系
mOb.subscribe(mSb);
}

private static Okhttp_Gson_Pager_Data getData(){
Okhttp_Gson_Pager_Data obj = new Okhttp_Gson_Pager_Data();
Okhttp_Gson_Pager_Data.ErrorBean err = new Okhttp_Gson_Pager_Data.ErrorBean();
err.setErr_code(10001);
obj.setError(err);
//errorInfoSend ();
return obj;
}

private static void errorInfoSend (){
int i = 0;
i /= i;
}

private static void method2() {
//这是发送端
Observable<Okhttp_Gson_Pager_Data> mOb = Observable.just(getData());

//这是接收端
Action1<Okhttp_Gson_Pager_Data> onNextAction = new Action1<Okhttp_Gson_Pager_Data>() {
@Override
public void call(Okhttp_Gson_Pager_Data obj) {
System.out.println("hello 我是Action1的接收端call方法！" + " 发送端发送的数据：" + obj.getError().getErr_code());
}
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
@Override
public void call(Throwable e) {
System.out.println("Action1<Throwable> onErrorAction --call(Throwable e) ：");
e.printStackTrace();
}
};
Action0 onCompleteAction = new Action0() {
@Override
public void call() {
System.out.println("Action0 onCompleteAction --  call() ：");
}
};

//建立联系
mOb.subscribe(onNextAction,onErrorAction,onCompleteAction);
}

private static void method3() {
//发送端
Observable
//发送数据
.just(getData())
//map是个输出类型转换函数
.map(new Func1<Okhttp_Gson_Pager_Data, Integer>() {
@Override
public Integer call(Okhttp_Gson_Pager_Data okhttp_gson_pager_data) {
return okhttp_gson_pager_data.getError().getErr_code();
}
})
//map是个输出类型转换函数
.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
return integer + "";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println("hello 我是Action1的接收端call方法！" + " 发送端发送的数据：" + s);
}
});
}

/**lift 方法复杂版*/
private static void method4(){
//发送器
OnSubscribe mObOnSub = new OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("hello！-- Observable.OnSubscribe mObOnSub = new Observable.OnSubscribe<String>--");
}
};

//原始发送端
Observable<String> originObservable =  Observable.create(mObOnSub);

Operator operator = new Operator<Integer,String>(){
@Override
public Subscriber<? super String> call(final Subscriber<? super Integer> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
int value = Integer.valueOf(s);
subscriber.onNext(value);
}
};
}
};

//目标发送端
Observable targetObservable = originObservable.lift(operator);

targetObservable.subscribe(new Action1<Integer>() {
@Override
public void call(Integer obj) {
System.out.println(obj);
}
});
}

/***
* 将String类型的值转化为Integer类型的值乘以2并返回
* lift 简单版
*/
private static void method5() {
//发送端
Observable
//发送器
.just("123")
//转换发送端
.lift(new Operator<Integer, String>() {
@Override
public Subscriber<? super String> call(final Subscriber<? super Integer> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
int value = Integer.valueOf(s);
value *= 2;
subscriber.onNext(value);
}
};
}
})
//转换发送器
.lift(new Operator<Integer, Integer>() {
@Override
public Subscriber<? super Integer> call(final Subscriber<? super Integer> subscriber) {
return new Subscriber<Integer>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(Integer integer) {
integer *= 100;
subscriber.onNext(integer);
}
};
}
})
//转换输出类型器
.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
return integer + "";
}
})
//接收端
.subscribe(new Action1<String>() {
//接收器
@Override
public void call(String obj) {
System.out.println("最后一个接收器 - onNext: 返回结果:" + obj);
}
});
}

private static Observable<String> liftAll(Observable observable) {
return observable
//转换发送端1
.lift(new Operator<String, String>() {
@Override
public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
subscriber.onNext(Integer.valueOf(s) * 5 + "");
}
};
}
})
//转换发送端2
.lift(new Operator<String, String>() {
@Override
public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
subscriber.onNext(Integer.valueOf(s) * 5 + "");
}
};
}
})
//转换发送端3
.lift(new Operator<String, String>() {
@Override
public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
subscriber.onNext(Integer.valueOf(s) * 5 + "");
}
};
}
})
//转换发送端4
.lift(new Operator<String, String>() {
@Override
public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
return new Subscriber<String>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}

@Override
public void onError(Throwable e) {
subscriber.onError(e);
}

@Override
public void onNext(String s) {
subscriber.onNext(Integer.valueOf(s) * 5 + "");
}
};
}
});
}

/**
* 升级版liftall使用
*/
private static void  method6() {
liftAll(Observable.just("123")).subscribe(new Action1() {
@Override
public void call(Object obj) {
System.out.println("最后一个接收器 - onNext: 返回结果:" + obj);
}
});
}

/**
**
*/
private static class LiftAllTransformer implements Observable.Transformer<Integer,String>{
@Override
public Observable<String> call(Observable<Integer> integerObservable) {
return liftAll(integerObservable);
}
}

/**
** compose 包装多次lift操作，进行发送端转换器集合复用器
*/
private static void method7() {
//发送端转换器 包装器
Observable.Transformer liftAll = new LiftAllTransformer();
//发送端
Observable
//发送器
.just("123")
//发送端转换器 包装器
.compose(liftAll)
//建立订阅关系
.subscribe(
//接收端
new Action1() {
//接收器
@Override
public void call(Object obj) {
System.out.println("最后一个接收器 - onNext: 返回结果:" + obj);
}
});
}

/**
** Scheduler 线程使用
* subscribeOn(Schedulers.xxxx()) 作用于前边的发送器，在数据开始发送的时候，就已经产生作用环境
* .observeOn(AndroidSchedulers.mainThread() / Schedulers.xxxx()) 产生影响在接收端（Subscriber )的接收器前边
*/
private static void method8() {
//发送端
Observable
//数组发送器
.from(new Integer[]{1,2,3,4})
//发送器 在IO线程发送，比较耗时的操作都在发送器操作
//.subscribeOn(Schedulers.io())
//下一个接收器  可以在一个新的线程中使用
//.observeOn(Schedulers.newThread())
//转换分解器 分解成一个个发送端
.flatMap(new Func1<Integer, Observable<String>>() {
//分解转换过后，返回一个发送端
@Override
public Observable<String> call(Integer integer) {
return Observable.just(integer + "");
}
})
//下一个接收器  在IO线程中使用
//.observeOn(Schedulers.io())
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return Integer.valueOf(s);
}
})
//下一个接收器  在主线程线程中使用，也即是UI可以更新的线程
//.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer obj) {
System.out.println("最后一个接收器打印所有结果 - onNext: 返回结果:" + obj);
}
});
}

/**
** doOnSubscribe 这个方法发生在发送器（OnSubscribe、just、from）方法之前
* 作用，可以进行设置进度条的UI操作（android 端操作页面的需求）
*/
private static void method9() {
//发送端
Observable
//发送器
.just("123")
//发送前执行器
.doOnSubscribe(new Action0() {
@Override
public void call() {
System.out.println("发送器前执行器的结果:" + "hello，my first excute method doOnSubscribe!");
}
})
//上一个数据发送器，工作的环境
//.subscribeOn(AndroidSchedulers.mainThread())
//建立关系
.subscribe(
//接收端
new Action1<String>() {
//接收器
@Override
public void call(String s) {
System.out.println("最后一个接收器 - onNext: 返回结果:" + s);
}
});
}

public static void main(String args[]) {
//        method1();
//        method2();
//        method3();
//        method4();
//        method5();
//        method6();
//        method7();
//        method8();
method9();
}
}

学习参考博文：http://gank.io/post/560e15be2dca930e00da1083#toc_12