RX系列三 | RxJava | create | from | interval | just | range | filter

简介: RX系列三 | RxJava | create | from | interval | just | range | filter我们在第一篇里有说过一些基本的关系,现在我们需要用到一些依赖,这里记得添加,我们本章就来看下他的执行顺序和一些基本的操作符,操作符是我们一定要去摸清楚的,是很重要的知识点,操作符的作用是很大的 compile 'io.

RX系列三 | RxJava | create | from | interval | just | range | filter


我们在第一篇里有说过一些基本的关系,现在我们需要用到一些依赖,这里记得添加,我们本章就来看下他的执行顺序和一些基本的操作符,操作符是我们一定要去摸清楚的,是很重要的知识点,操作符的作用是很大的

    compile 'io.reactivex:rxjava:1.1.0'
    compile 'io.reactivex:rxandroid:1.1.0'

首先,我们要明白,他的执行步骤是什么

执行步骤

与其说执行步骤,倒不如说执行流程,在RxJava中,我们首先,会创建一个Observable,也就是会观察者,去执行一定的逻辑,然后定义一个观察者Subscriber去接收结果,最后,通过

    observable.subscribe(showSub);

去订阅,而且很有意思的是,因为Rx的链式语法,我们可用一直next下去,这样,我们先来看一段简短的例子,你就明白大概的意思了,比如我现在要执行一个这样的操作

   //创建
    private void rxJavaCreate() {
        //定义被观察者
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //判断是否有订阅的关系
                if (!subscriber.isUnsubscribed()) {
                    subscriber.onNext("Hello");
                    subscriber.onNext("Hi");
                    subscriber.onNext(getJson());
                    subscriber.onNext("END");
                    //完成
                    subscriber.onCompleted();
                }
            }
        });

        Subscriber<String> showSub = new Subscriber<String>() {
            //结束
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted");
            }

            //发生错误
            @Override
            public void onError(Throwable e) {
                Log.i(TAG, e.toString());
            }

            //继续执行
            @Override
            public void onNext(String s) {
                Log.i(TAG, s);
            }
        };
        //订阅
        observable.subscribe(showSub);
    }

    public static String getJson() {
        //这里进行网络解析 耗时操作
        return "Json Data";
    }

这段代码不难,你仔细去读,你就会发现,其实,这段代码就四个部分,首先是一个Observable,然后是一个Subscriber,接着就是订阅了。当然还有一个getJson的方法,那我们来具体分析一下这段代码吧,了解了这段代码,也就对RxJava有了一个皮毛的理解了

首先我们要看下Observable,他需要定义一个泛型的类型参数,默认是Object,这里我定义的是String,通过Observable.create去创建Observable对象,并且重写他里面的call方法,我们就可以在里面进行我们想要的操作了,先看下里面的这段代码

    //判断是否有订阅的关系
    if (!subscriber.isUnsubscribed()) {
         subscriber.onNext("Hello");
         subscriber.onNext("Hi");
         subscriber.onNext(getJson());
         subscriber.onNext("END");
         //完成
         subscriber.onCompleted()
     }

这段代码可以看出,我们首先来判断isUnsubscribed是否存在订阅关系,然后去执行,这里先是一个Hello,然后时一个Hi,接着调用getJson,最后END,看起来是这样,但是我要告诉你的是,如果getJson是一个网络请求的耗时操作,那他们的执行顺序是什么呢?其实他们还是顺序执行,最后调用的是subscriber.onCompleted()来告诉Subscriber,我已经OK了,那我们再看Subscriber就比较好说话了,他里面三个方法都不用多说了吧,一个是结束,一个是错误,还有一个继续执行,最后通过订阅就行,那我们来看下他真的执行打印顺序是什么呢?

这里写图片描述

可以看到,的确是顺序执行,最后,调用onCompleted的回调,流程算是清楚了吧,但是其实我们并没有讲真正的使用,不过在此之前,还是要说一下,RxJava的一个特点就是链式,也就是简洁,特别是配合了lambda表达式,那就更爽了,不过这个我们后面讲,先看一下传统的缩写是怎样的

    //链式表达
    private void  rxJavaCreateSmall(){
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 10; i++) {
                    subscriber.onNext(i);
                }
                //结束
                subscriber.onCompleted();
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, e.toString());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, integer.toString());
            }
        });
    }

这样也是可以的,可以看到,我们执行了十次的onNext,看下并且这是一个链式表达,这就是一个整个的流程了,我们来看下执行结果:

这里写图片描述

from操作符

其实这里还有一种比较简单的方式就是用from操作符了,操作符和我们是密不可分的,所以我这里先提及一下,首先说一下from操作符的含义

  • 将其他种类的对象和数据类型转换为Observable,我们看一段代码来理解他
   //from操作符
    private void fromOperation(){
        Integer [] items = {1,2,3,4,5};
        //使用在被观察者,返回的对象一般都是数值类型
        Observable observable =Observable.from(items);
        observable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                Log.i(TAG,o.toString());
            }
        });
    }

当你使用Observable时,如果你要处理的数据都可以转换成为Observables,而不是需要混合使用Observables和其他类型的数据

使用在被观察者,返回的对象一般都是数值类型

这里写图片描述

interval操作符

form就是顺序的发送数据,其实还有一个操作符,也是可以的,那就是interval,指定某一时刻进行数据发送,有点类似与定时器,我们来看下这段代码

  //指定某一时刻进行数据发送
    private void intervalOperation(){
        //每隔一秒发送数据
        Observable observable =Observable.interval(1,1, TimeUnit.SECONDS);
        observable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                Log.i(TAG,o.toString());
            }
        });
    }

这段代码就是每隔一秒发送一个数据,我们默认是累加,主要还是在interval这个参数里面,我们看下效果

这里写图片描述

just操作符

前面两个都是处理整形的,但是如果你需要处理数组集合呢?这个时候你就需要用到just了,我们来看下他的实例代码

   //just操作符 处理数组集合
    private void justOperation(){
        Integer [] items1 = {1,2,3,4,5};
        Integer [] items2 = {6,7,8,9,10};
        Observable observable =Observable.just(items1,items2);
        observable.subscribe(new Subscriber<Integer[]>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, e.toString());
            }

            @Override
            public void onNext(Integer[] i) {
                //这里是数组输出
                for (int j = 0; j < i.length; j++) {
                    Log.i(TAG, i[j].toString());
                }
            }
        });
    }

这里我们就是打印item1和item2的数据了,当然,我这里为什么传两个,你传一个也可以嘛,这只是一个参数而已,我们来看下他的运行结果

这里写图片描述

这里可以看到的是顺序执行,先item1再item2了

range操作符

其实在上面,还有一种没有说到,那就是指定范围内输出,就像interval一样,他是无限的累加,但是我现在输出的返回控制在一定的数据内中,这就要使用range了,我们来看下代码

     //范围输出
    private void rangeOperation(){
        Observable observable =Observable.range(1,10);
        observable.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, e.toString());
            }

            @Override
            public void onNext(Integer i) {
                Log.i(TAG, i.toString());
            }
        });
    }

这里我指定输出范围是1-10,那么他执行的结果又会是什么呢?

这里写图片描述

可以看到,输出到10就没有再输出了

filter操作符

有范围输出,那肯定有过滤了,越看越觉得和String的方法有点像,但是你别忘记了,RxJava有一个特点就是异步,那里才是真正的有用的地方,现在我只是先把操作符的实例告诉大家,但是如果还没有运用到实际项目中去的话,你应该还属于半懵逼的状态,不过不要紧,我们继续看:说一下filter,他是过滤器,我们直接来看代码

    //过滤
    private void filterOperation(){
        Observable.just(1,2,3,4,5,6,7,8,9,10).filter(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) {
                return integer < 5;
            }
            //表示通过网络获取数据
        }).observeOn(Schedulers.io()).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, e.toString());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, integer.toString());
            }
        });
    }

这里我就直接链式了,我们可以看到Observable.just处理一个数组,向观察者传递数据,然后观察者拿到数据之后过滤一下, return integer < 5,只有小于5的数据才会被处理,再接着订阅,所以可以看到,输出的结果就是这个样子了

这里写图片描述

像RxJava的操作符是有很多的,但是这里就不一一列举了,还是需要大家自己去理解,带个路就行,后面结合实例,我们再来回过头来看这些例子,你就明白了,这里我也找到一个RxJava操作符的一览,有兴趣的可以看下:

当然,你如果有兴趣的话,可以找张梯子就看一下JW大神的视频讲解:JW大神的视频教学

OK,我这一章就先到这里了,有兴趣的可以接着往下看

Sample下载:系列最后一篇提供

有兴趣的加群:555974449

目录
相关文章
|
3月前
|
机器学习/深度学习 关系型数据库 Ruby
GitlabCI学习笔记之三:GitLabRunner pipeline语法之tags allow_faillure when retry timeout parallel
GitlabCI学习笔记之三:GitLabRunner pipeline语法之tags allow_faillure when retry timeout parallel
|
数据库
解决numpy.core._exceptions.UFuncTypeError: ufunc ‘add‘ did not contain a loop with signature matching
解决numpy.core._exceptions.UFuncTypeError: ufunc ‘add‘ did not contain a loop with signature matching
1093 0
解决numpy.core._exceptions.UFuncTypeError: ufunc ‘add‘ did not contain a loop with signature matching
AMQPProtocolChannelException: PRECONDITION_FAILED - inequivalent arg ‘x-message-ttl‘ for queue ‘norm
AMQPProtocolChannelException: PRECONDITION_FAILED - inequivalent arg ‘x-message-ttl‘ for queue ‘norm
112 0
|
索引
RxSwift操作符take、skip、materialize、withLatestFrom、interval等的使用
RxSwift操作符take、skip、materialize、withLatestFrom、interval等的使用
246 1
Sap Ds Data is not available. Increase the time-out interval values in Debug | Options
Sap Ds Data is not available. Increase the time-out interval values in Debug | Options
134 0
nvprof --query-events
nvprof --query-events
116 0
|
NoSQL
RxJava/RxAndroid:ConnectableObservable &amp; replay(int bufferSize, long time, TimeUnit unit)
RxJava/RxAndroid:ConnectableObservable & replay(int bufferSize, long time, TimeUnit unit) import android.
1062 0
|
Android开发
RxJava/RxAndroid:ConnectableObservable &amp; replay(long time, TimeUnit unit)
RxJava/RxAndroid:ConnectableObservable & replay(long time, TimeUnit unit) import android.
1094 0
RxJava/RxAndroid:ConnectableObservable &amp; replay(int bufferSize)
RxJava/RxAndroid:ConnectableObservable & replay(int bufferSize) import android.
940 0