RxAndroid 主/子线程切换配合的三种方式

Myth丶恋晨 2022-07-16 04:26 188阅读 0赞

我使用RxJava和RxAndroid的一个主要目的就是为了让逻辑复杂的业务需求在代码表现上不会特别混乱，以前在接手别人的项目的时候，经常碰到if else嵌套好几层的情况，还有当同一页面上出现好多异步任务的时候，会出现大量的回调的嵌套，程序员往往分了很多的函数东一笔西一笔的，读起来非常困难。

而RxJava解决这类问题就很得心应手，RxJava可以把复杂的业务逻辑用一条线串连起来，没有复杂的回调嵌套，也不会出现大型项目中常见的“谜之缩进”的代码，更可贵的是RxJava在处理异常上比传统try catch好的多，让代码更整洁和逻辑清晰。尤其在一项业务涉及到好多个网络接口，数据库查询，和一些耗时的计算操作，非常适合使用RxJava的方式将主线程和子线程执行的动作串联在一张页面上，通过自上而下的阅读代码就可以理清楚整个业务流程，对于大型项目来说非常实用。

这里写一个小demo来演示RxAndroid在碰到很多异步任务的情况下，在主线程和子线程之间灵活切换的案例。

比如现在有这样一个需求，给出一个url，如http://qduxsh.zz.vc/html/article.php?id=33，取出这个url中的协议名“http”，域名“qduxsh.zz.vc”,和api地址“/html/article.php”和全部的get请求参数“id=33”

![Center][]

方式一：异步任务式调用

现在我们假设从url中解析出这些信息是一个非常耗时的操作，需要放在子线程中执行，并在全部结果均获取完毕后显示在UI上，那么使用RxJava可以像下面一样，使用有点像AsynTask的写法实现不同线程的异步回调

/**
         * 一个简单的封装类
         */
        public class UrlEntity{
            public String protocol;
            public String host;
            public String api;
            public String params;
        }

final String url = "http://qduxsh.zz.vc/html/article.php?id=33";
            Observable.create(new Observable.OnSubscribe<UrlEntity>() {
                @Override
                public void call(Subscriber<? super UrlEntity> subscriber) {//子线程执行的方法
                    Log.i("Alex", "call()执行线程是" + Thread.currentThread().getName() + "线程优先级=" + Thread.currentThread().getPriority() + "  线程id=" + Thread.currentThread().getId());
                    //假设一下几行处理都是耗时操作
                    Uri uri = Uri.parse(url);
                    String protocol = uri.getScheme();
                    String host = uri.getHost();
                    String api = uri.getPath();
                    String params = url.substring(url.lastIndexOf('?')+1,url.length());
                    UrlEntity entity = new UrlEntity();
                    entity.protocol = protocol;
                    entity.host = host;
                    entity.api = api;
                    entity.params = params;
                    //以上内容均在子线程执行
                    //现在开始触发主线程的回调
                    subscriber.onNext(entity);
    
                    subscriber.onCompleted();
                }
            })
                    .subscribeOn(Schedulers.computation()) // 指定 subscribe() 发生在 运算 线程
                    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
                    .subscribe(new Observer<UrlEntity>() {
    
                        @Override
                        public void onNext(UrlEntity entity) {//主线程执行的方法
                            Log.i("Alex", "onNext()执行线程是" + Thread.currentThread().getName() + "线程优先级=" + Thread.currentThread().getPriority() + "  线程id=" + Thread.currentThread().getId());
                            TextView tv_protocol = (TextView) findViewById(R.id.tv_protocol);
                            tv_protocol.setText(entity.protocol);
                            TextView tv_host = (TextView) findViewById(R.id.tv_host);
                            tv_host.setText(entity.host);
                            TextView tv_api = (TextView) findViewById(R.id.tv_api);
                            tv_api.setText(entity.api);
                            TextView tv_param_list = (TextView) findViewById(R.id.tv_param_list);
                            tv_param_list.setText(entity.params);
                        }
    
                        @Override
                        public void onCompleted() {
                            Log.i("Alex", "onCompleted");
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            Log.i("ALex", "出现了异常", e);
                            Toast.makeText(MainActivity.this, "Error!", Toast.LENGTH_SHORT).show();
                        }
                    });

那么现在肯定有人会说了，AsynTask也可以实现同样的效果，而且代码比这个更简洁易懂，那么我们还要RxJava干什么，但是假如我们现在修改一下需求，不要等协议，域名，api，参数都解析完毕才显示在UI上，而是这几个需求同时解析，谁先解析完谁就先显示，让用户尽快的看到页面的变化，那么用RxJava写应该如何呢？

方式二：子线程分步任务逐个回调

public enum Type {
            host,
            protocol,
            api,
            params
        }
        public static class KeyValue{
            public Type type;
            public String value;
            public KeyValue(Type type, String value) {
                this.type = type;
                this.value = value;
            }
        }
    
    Observable.create(new Observable.OnSubscribe<KeyValue>() {
                @Override
                public void call(Subscriber<? super KeyValue> subscriber) {//子线程执行的方法
                    Log.i("Alex", "call()执行线程是" + Thread.currentThread().getName() + "线程优先级=" + Thread.currentThread().getPriority() + "  线程id=" + Thread.currentThread().getId());
                    Uri uri = Uri.parse(url);
                    String protocol = uri.getScheme();//假设这是一个耗时操作
                    subscriber.onNext(new KeyValue(Type.protocol, protocol));
                    String host = uri.getHost();//假设这是一个耗时操作
                    subscriber.onNext(new KeyValue(Type.host, host));
                    String path = uri.getPath();//假设这是一个耗时操作
                    subscriber.onNext(new KeyValue(Type.api, path));
                    String params = url.substring(url.lastIndexOf('?')+1,url.length());
                    //现在触发回调
                    subscriber.onNext(new KeyValue(Type.params,params));
                    subscriber.onCompleted();
                }
            })
                    .subscribeOn(Schedulers.computation()) // 指定 subscribe() 发生在 运算 线程
                    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
                    .subscribe(new Observer<KeyValue>() {
    
                        @Override
                        public void onNext(KeyValue entry) {//主线程执行的方法
                            Log.i("Alex", "onNext()执行线程是" + Thread.currentThread().getName() + "线程优先级=" + Thread.currentThread().getPriority() + "  线程id=" + Thread.currentThread().getId());
                            switch (entry.type) {
                                case protocol:
                                    TextView tv_protocol = (TextView) findViewById(R.id.tv_protocol);
                                    tv_protocol.setText(entry.value);
                                    break;
                                case host:
                                    TextView tv_host = (TextView) findViewById(R.id.tv_host);
                                    tv_host.setText(entry.value);
                                    break;
                                case api:
                                    TextView tv_api = (TextView) findViewById(R.id.tv_api);
                                    tv_api.setText(entry.value);
                                    break;
                                case params:
                                    TextView tv_param_list = (TextView) findViewById(R.id.tv_param_list);
                                    tv_param_list.setText(entry.value);
                                    break;
                            }
                        }
    
                        @Override
                        public void onCompleted() {
                            Log.i("Alex", "onCompleted");
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            Log.i("ALex", "出现了异常", e);
                            Toast.makeText(MainActivity.this, "Error!", Toast.LENGTH_SHORT).show();
                        }
                    });

但是像上面那样写有很多的坏处  
首先，所有的url解析均在同一个线程排队执行，也就是说不解析完protocol没法去解析域名，不解析完域名也没法去解析api，这样执行的效率就很低了。

其次，上面所有的解析结果回调都进入onNext()方法，并在onNext()中使用switch来判断是哪个项目被解析完成，这样会导致onNext()这个函数特别巨大，如果我们对解析完毕的结果还要进行许多许多其他的处理，那么在onNext()方法中又会是一番混乱的景象，所以我们要在同一条Observerble链中进行4项异步的工作，并将耗时的操作全部放在子线程中执行，所有的UI操作放在主线程当中执行。

但是问题来了，在RxJava中，一个Observerble只能指定一个observerOn()线程池和一个subscribeOn()线程池，也就是说，如果我们需要在观察者中使用主线程回调，那么回调只能有一次，那就必须得等所有子线程均执行完以后统一回调，可是这样一来我们就无法实现完成一次解析显示一次结果的需求了。

方式三：Observerble嵌套实现多重异步任务

不过RxJava中给出的flatMap()可以解决这一点，flatMap()的左右就是将一个传入的任意类型的参数转换成Observerble对象的，虽然一个Observerble对象只能指定一个子线程池和一个回调线程池，但是我们可以通过flatMap()创建多个处理不同任务的Observerble，这样就实现了多次回调的效果，而且在RxJava中，我们可以方便的再加入一个url识别过滤的功能，代码如下

String[] urls = {
                    "192.168.1.1",
                    "http://qduxsh.zz.vc/html/article.php?id=33",
                    "/home/alex/a.jpg"
            };
            
            
    Observable.from(urls)
                    .filter(new Func1<String, Boolean>() {
                        @Override
                        public Boolean call(String s) {//子线程执行
                            Log.i("Alex","filter线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                            if(s.startsWith("http"))return true;
                            return false;
                        }
                    })
                    .map(new Func1<String, String>() {
                        @Override
                        public String call(String s) {//子线程执行
                            Log.i("Alex","map线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                            Log.i("Alex","通过过滤的url是：："+s);
                            //在这里可以对过滤完的url稍作处理,比如加个参数
                            s = s+"&name=alex";
                            return s;
                        }
                    })
                    .subscribeOn(Schedulers.computation()) // 指定 subscribe() 发生在 运算 线程
                    //因为一个observerble的内部只能设置一个subscribeOn的线程，所以要使用flatMap方法创建多个observerble，就可以实现多个子线程和主线成的配合
                    .flatMap(new Func1<String, Observable<String>>() {//这是获取url中协议名称的一个Observerble
                        @Override
                        public Observable<String> call(final String s) {//子线程执行
                            Observable o = Observable.create(new Observable.OnSubscribe<String>() {
                                @Override
                                public void call(Subscriber<? super String> subscriber) {
                                    Log.i("Alex","flatmap传入的url是"+s);
                                    Log.i("Alex","flatmap call 线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                    //现在开始获取url的协议名称
                                    Uri uri = Uri.parse(url);
                                    String protocol = uri.getScheme();//假设这是一个耗时操作
                                    subscriber.onNext(protocol);
                                    subscriber.onCompleted();
                                }
                            });
                            o.subscribeOn(Schedulers.computation())//让获取url协议的代码在子线程中执行
                                    .observeOn(AndroidSchedulers.mainThread())//让获取成功后的回调在主线程中执行
                                    .subscribe(new Observer<String>() {//主线程的回调
                                        @Override
                                        public void onCompleted() {
    
                                        }
    
                                        @Override
                                        public void onError(Throwable e) {
    
                                        }
    
                                        @Override
                                        public void onNext(String protocol) {
                                            Log.i("Alex","flatmap的onnext线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                            Log.i("Alex","flatmap的onnext线程收到的参数是"+protocol);
                                            TextView tv_protocol = (TextView) findViewById(R.id.tv_protocol);
                                            tv_protocol.setText(protocol);
                                        }
                                    });
                            return o;
                        }
                    })
                    .flatMap(new Func1<String, Observable<String>>() {//这是获取url中域名的一个Observerble
                        @Override
                        public Observable<String> call(final String s) {
                            Observable o = Observable.create(new Observable.OnSubscribe<String>() {
                                @Override
                                public void call(Subscriber<? super String> subscriber) {
                                    Log.i("Alex","flatmap传入的url是"+s);
                                    Log.i("Alex","flatmap call 线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                    //现在开始获取url的协议名称
                                    Uri uri = Uri.parse(url);
                                    String host = uri.getHost();//假设这是一个耗时操作
                                    subscriber.onNext(host);
                                    subscriber.onCompleted();
                                }
                            });
                            o.subscribeOn(Schedulers.computation())//让获取url协议的代码在子线程中执行
                                    .observeOn(AndroidSchedulers.mainThread())//让获取成功后的回调在主线程中执行
                                    .subscribe(new Observer<String>() {
                                        @Override
                                        public void onCompleted() {
    
                                        }
    
                                        @Override
                                        public void onError(Throwable e) {
    
                                        }
    
                                        @Override
                                        public void onNext(String host) {
                                            Log.i("Alex","flatmap的onnext线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                            Log.i("Alex","flatmap的onnext线程收到的参数是"+host);
                                            TextView tv_host = (TextView) findViewById(R.id.tv_host);
                                            tv_host.setText(host);
                                        }
                                    });
                            return o;
                        }
                    })
                    .flatMap(new Func1<String, Observable<String>>() {//这是获取url中api的一个Observerble
                        @Override
                        public Observable<String> call(final String s) {
                            Observable<String> o = Observable.just(s);
                            o.map(new Func1<String, String>() {//子线程执行
                                @Override
                                public String call(String s) {
                                    Log.i("Alex","flatmap传入的url是"+s);
                                    Log.i("Alex","flatmap call 线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                    //现在开始获取url的协议名称
                                    Uri uri = Uri.parse(url);
                                    String api = uri.getPath();//假设这是一个耗时操作
                                    return api;
                                }
                            })
                                    .subscribeOn(Schedulers.computation())//让获取url协议的代码在子线程中执行
                                    .observeOn(AndroidSchedulers.mainThread())//让获取成功后的回调在主线程中执行
                                    .subscribe(new Action1<String>() {//主线程执行
                                        @Override
                                        public void call(String s) {
                                            Log.i("Alex","flatmap的onnext线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                            Log.i("Alex","flatmap的onnext线程收到的参数是"+s);
                                            TextView tv_api = (TextView) findViewById(R.id.tv_api);
                                            tv_api.setText(s);
                                        }
                                    });
                            return o;
                        }
                    })
                    .flatMap(new Func1<String, Observable<String>>() {//这是获取url中参数的一个Observerble
                        @Override
                        public Observable<String> call(final String s) {
                            Observable<String> o = Observable.just(s);
                            o.map(new Func1<String, String>() {//子线程执行
                                @Override
                                public String call(String s) {
                                    Log.i("Alex","flatmap传入的url是"+s);
                                    Log.i("Alex","flatmap call 线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                    //现在开始获取url的协议名称
                                    String params = url.substring(url.lastIndexOf('?')+1,url.length());
                                    return params;
                                }
                            })
                                    .subscribeOn(Schedulers.computation())//让获取url参数的代码在子线程中执行
                                    .observeOn(AndroidSchedulers.mainThread())//让获取成功后的回调在主线程中执行
                                    .subscribe(new Action1<String>() {//主线程执行
                                        @Override
                                        public void call(String s) {
                                            Log.i("Alex","flatmap的onnext线程是"+Thread.currentThread().getName()+"线程优先级="+Thread.currentThread().getPriority()+"  线程id="+Thread.currentThread().getId());
                                            Log.i("Alex","flatmap的onnext线程收到的参数是"+s);
                                            TextView tv_param_list = (TextView) findViewById(R.id.tv_param_list);
                                            tv_param_list.setText(s);
                                        }
                                    });
                            return o;
                        }
                    })
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Action1<String>() {
                        @Override
                        public void call(String s) {
                            Log.i("Alex","所有任务均以布置完毕");
                        }
                    });

从这里我们可以看出，使用RxJava以后虽然我们用一条链的形式串联起了一个多重回调的需求，让逻辑变得线性并且易读，但是代码长度仍然很可观。但是RxJava本身也给出了很多缩减代码长度的接口，比如上面的代码中，我获取协议名称，域名两端代码要长于获取api和请求参数的两段代码，后两段代码不仅程度有所减少，最重要的是减少了没用到的抽象函数的出现频率，让代码更加整洁。

这种用纵向空间取代函数嵌套的写法也许并不是所有人都接收，但是我们对比一下同样一段逻辑使用AsynTask的写法就能体会出来两者的差异，因为多重回调的嵌套非常容易出现“谜之缩进”的情况

new AsyncTask<String,Void,String>(){
    
                @Override
                protected String doInBackground(String... params) {
                    try {
                        Uri uri = Uri.parse(params[0]);
                        String protocol = uri.getScheme();//假设这是一个耗时操作
                        return protocol;
                    }catch (Exception e){
                        return "";
                    }
    
                }
    
                @Override
                protected void onPostExecute(String s1) {
                    super.onPostExecute(s1);
                    TextView tv_protocol = (TextView) findViewById(R.id.tv_protocol);
                    tv_protocol.setText(s1);
                    new AsyncTask<String,Void,String>(){
    
                        @Override
                        protected String doInBackground(String... params) {
                            try {
                                Uri uri = Uri.parse(params[0]);
                                String host = uri.getHost();//假设这是一个耗时操作
                                return host;
                            }catch (Exception e){
                                return "";
                            }
    
                        }
    
                        @Override
                        protected void onPostExecute(String s2) {
                            super.onPostExecute(s2);
                            try{
                                TextView tv_host = (TextView) findViewById(R.id.tv_host);
                                tv_host.setText(s2);
                            }catch (Exception e){
    
                            }
    
                            new AsyncTask<String,Void,String>(){
    
                                @Override
                                protected String doInBackground(String... params) {
                                    try {
                                        Uri uri = Uri.parse(params[0]);
                                        String path = uri.getPath();//假设这是一个耗时操作
                                        return path;
                                    }catch (Exception e){
                                        return "";
                                    }
    
                                }
    
                                @Override
                                protected void onPostExecute(String s) {
                                    super.onPostExecute(s);
                                    try {
                                        TextView tv_api = (TextView) findViewById(R.id.tv_api);
                                        tv_api.setText(s);
                                    }catch (Exception e) {
    
                                    }
                                    new AsyncTask<String,Void,String>(){
    
                                        @Override
                                        protected String doInBackground(String... params) {
                                            try {
                                                String parameters = params[0].substring(params[0].lastIndexOf('?')+1,params[0].length());
                                                return parameters;
                                            }catch (Exception e){
                                                return "";
                                            }
                                        }
    
                                        @Override
                                        protected void onPostExecute(String s) {
                                            super.onPostExecute(s);
                                            try {
                                                TextView tv_param_list = (TextView) findViewById(R.id.tv_param_list);
                                                tv_param_list.setText(s);
                                            }catch (Exception e){
    
                                            }
    
                                        }
                                    }.execute(url);
                                }
                            }.execute(url);
                        }
                    }.execute(url);
                }
            }.execute(url);

上面那段代码使用了大量的缩进，再加上try catch的包裹，让代码显得非常臃肿，而实现同样逻辑的RxJava代码，可以看我上面贴出的第二部分，在可读性上好更好一点，所以在处理这种链式逻辑的时候，我更倾向使用RxJava多一点

[Center]: /images/20220716/0ce638782751496b99156072c4e9bedd.png