andorid jar/库源码解析之RxJava2

目录：andorid jar/库源码解析 

rxjava2：

　　作用：

　　　　通过提供一种，观察者和订阅者的模式，的架构，来优化逻辑流程。适用于复杂和需要多数据转换和长流程。

　　栗子：

 　　定义三个对象类

public class resultinfo {
    public int code;
    public string msg;
    public string data;
}

public class userinfo {
    public int status;
    public string name;
    public string head;
    public list<skillinfo> skillinfolist;
}

public class skillinfo {
    public string name;
    public int level;

    public skillinfo(string name, int level){
        this.name = name;
        this.level = level;
    }
}

　　一段逻辑测试代码：

    private resultinfo login_http(string name, string pwd){
        resultinfo resultinfo = new resultinfo();
        resultinfo.code = 0;
        resultinfo.msg = "";

        gson gson = new gson();
        userinfo userinfo = new userinfo();
        userinfo.status = 1;
        userinfo.name = "";
        userinfo.head = "";
        userinfo.skillinfolist = new arraylist<>();
        userinfo.skillinfolist.add(new skillinfo("音乐", 10));
        userinfo.skillinfolist.add(new skillinfo("美术", 6));
        userinfo.skillinfolist.add(new skillinfo("体育", 9));
        resultinfo.data = gson.tojson(userinfo);

        return resultinfo;
    }

　　版本1：（定义一个被观察的对象，和订阅者，设置他们使用的线程，最后通过调用 subscribe，使他们关联起来。且执行）

observable<resultinfo> observable = observable.create(new observableonsubscribe<resultinfo>() {
            @override
            public void subscribe(observableemitter<resultinfo> e) throws exception {
                e.onnext(login_http(name, pwd));
            }
        });

        consumer<resultinfo> consumer = new consumer<resultinfo>() {
            @override
            public void accept(resultinfo resultinfo) throws exception {
                if(resultinfo.code == 0){
                    gson gson = new gson();
                    userinfo userinfo = gson.fromjson(resultinfo.data, userinfo.class);
                    if(userinfo.status == 0){
                        // 注册
                        toast.maketext(getbasecontext(), "注册", toast.length_long).show();
                    } else if(userinfo.status == 1){
                        // 登录
                        toast.maketext(getbasecontext(), "登录", toast.length_long).show();
                    }
                }
            }
        };

        // subscribeon() 指定的是发送事件的线程, observeon() 指定的是接收事件的线程.
        observable.subscribeon(schedulers.newthread())
                .observeon(androidschedulers.mainthread())
                .subscribe(consumer);

　　版本2：（在版本1的基础上，通过在被观察者调用map处理，观察的数据，使数据只返回结果，而订阅者也是对结果进行判断，这里使用的数据类型发生了改变）。

observable<integer> observable = observable.create(new observableonsubscribe<resultinfo>() {
            @override
            public void subscribe(observableemitter<resultinfo> e) throws exception {
                e.onnext(login_http(name, pwd));
            }
        }).map(new function<resultinfo, integer>() {
            @override
            public integer apply(resultinfo resultinfo) throws exception {
                return resultinfo.code;
            }
        });

        consumer<integer> consumer = new consumer<integer>() {
            @override
            public void accept(integer code) throws exception {
                if(code == 0){
                    toast.maketext(getbasecontext(), "成功", toast.length_long).show();
                } else{
                    toast.maketext(getbasecontext(), "失败", toast.length_long).show();
                }
            }
        };

        // subscribeon() 指定的是发送事件的线程, observeon() 指定的是接收事件的线程.
        observable.subscribeon(schedulers.newthread())
                .observeon(androidschedulers.mainthread())
                .subscribe(consumer);

　　版本3：（使用rxjava的优势，链式调用，来完成逻辑）

observable.create(new observableonsubscribe<resultinfo>() {
            @override
            public void subscribe(observableemitter<resultinfo> e) throws exception {
                e.onnext(login_http(name, pwd));
            }
        }).map(new function<resultinfo, integer>() {
            @override
            public integer apply(resultinfo resultinfo) throws exception {
                return resultinfo.code;
            }
        }).subscribeon(schedulers.newthread())
                .observeon(androidschedulers.mainthread())
                .subscribe(new consumer<integer>() {
                    @override
                    public void accept(integer code) throws exception {
                        if(code == 0){
                            toast.maketext(getbasecontext(), "成功", toast.length_long).show();
                        } else{
                            toast.maketext(getbasecontext(), "失败", toast.length_long).show();
                        }
                    }
                });

　　版本4：（同时引入了flatmap来，处理集合数据，返回集合，每个数据都可以响应订阅者的accept方法，同时引入了filter对数据进行过滤，类似于c#中的，linq用法。）

observable.create(new observableonsubscribe<resultinfo>() {
            @override
            public void subscribe(observableemitter<resultinfo> e) throws exception {
                e.onnext(login_http(name, pwd));
            }
        }).map(new function<resultinfo, userinfo>() {
            @override
            public userinfo apply(resultinfo resultinfo) throws exception {
                userinfo userinfo = new gson().fromjson(resultinfo.data, userinfo.class);

                return userinfo;
            }
        }).flatmap(new function<userinfo, observable<skillinfo>>() {
            @override
            public observable<skillinfo> apply(userinfo userinfo) throws exception {
                return observable.fromarray(userinfo.skillinfolist.toarray(new skillinfo[userinfo.skillinfolist.size()]));
            }
        }).filter(new predicate<skillinfo>() {
            @override
            public boolean test(skillinfo skillinfo) throws exception {
                return skillinfo.level > 8;
            }
        })
        .subscribeon(schedulers.newthread())
                .observeon(androidschedulers.mainthread())
                .subscribe(new consumer<skillinfo>() {
                    @override
                    public void accept(skillinfo skillinfo) throws exception {
                        toast.maketext(getbasecontext(), skillinfo.name + " " + skillinfo.level, toast.length_short).show();
                    }
                });

        // 类似于linq , take, skip deng

　　版本5：当然，rxjava的用法不仅仅这些，还有没有提到的，take，取几个成员，skip,跳过几个成员。等，这些用法和c#中的linq类似，详细很好理解

　　// 代码，略

　　源码解读：

　　// 源码解读，后面再补上。

　　源码：https://github.com/reactivex/rxjava

　　引入：

implementation "io.reactivex.rxjava2:rxjava:2.0.7"
implementation "io.reactivex.rxjava2:rxandroid:2.0.1"