zoukankan      html  css  js  c++  java
  • RxJava 和 RxAndroid 二(操作符的使用)

    前言:对Rx不了解的朋友可以先看我的第一篇博文 RxJava 和 RxAndroid 一 (基础),是对Rxjava的基本介绍

    1、merge操作符,合并观察对象

    复制代码
    19         List<String> list1 = new ArrayList<>() ;
    20         List<String> list2 = new ArrayList<>() ;
    21 
    22         list1.add( "1" ) ;
    23         list1.add( "2" ) ;
    24         list1.add( "3" ) ;
    25 
    26         list2.add( "a" ) ;
    27         list2.add( "b" ) ;
    28         list2.add( "c" ) ;
    29 
    30         Observable observable1 = Observable.from( list1 ) ;
    31         Observable observable2 = Observable.from( list2 ) ;
    32 
    33         //合并数据  先发送observable2的全部数据,然后发送 observable1的全部数据
    34         Observable observable = Observable.merge( observable2 , observable1 ) ;
    35 
    36         observable.subscribe(new Action1() {
    37             @Override
    38             public void call(Object o) {
    39              System.out.println( "rx-- " + o );
    40             }
    41         }) ;
    42 
    复制代码

    运行结果

    2、zip  操作符,合并多个观察对象的数据。并且允许 Func2()函数重新发送合并后的数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    List<String> list1 = new ArrayList<>() ;
           List<String> list2 = new ArrayList<>() ;
     
           list1.add( "1" ) ;
           list1.add( "2" ) ;
           list1.add( "3" ) ;
     
           list2.add( "a" ) ;
           list2.add( "b" ) ;
           list2.add( "c" ) ;
           list2.add( "d" ) ;
     
           Observable observable1 = Observable.from( list1 ) ;
           Observable observable2 = Observable.from( list2 ) ;
            
          Observable observable3 =  Observable.zip(observable1, observable2, new Func2<String , String , String >() {
               @Override
               public String call(String s1 , String s2 ) {
                   return s1 + s2  ;
               }
           }) ;
     
           observable3.subscribe(new Action1() {
               @Override
               public void call(Object o) {
                   System.out.println( "zip-- " + o );
               }
           }) ;

      运行效果:从效果图上可以看出,合并两个的观察对象数据项应该是相等的;如果出现了数据项不等的情况,合并的数据项以最小数据队列为准。

    3、scan累加器操作符的使用

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5  ) ;
           observable.scan(new Func2<Integer,Integer,Integer>() {
               @Override
               public Integer call(Integer o, Integer o2) {
                   return o + o2 ;
               }
           })
             .subscribe(new Action1() {
                 @Override
                 public void call(Object o) {
                     System.out.println( "scan-- " +  o );
                 }
             })   ;

      运行效果:

            第一次发射得到1,作为结果与2相加;发射得到3,作为结果与3相加,以此类推,打印结果:

           

    4、filter 过滤操作符的使用

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
    observable.filter(new Func1<Integer , Boolean>() {
        @Override
        public Boolean call(Integer o) {
            //数据大于4的时候才会被发送
            return o > 4 ;
        }
    })
            .subscribe(new Action1() {
                @Override
                public void call(Object o) {
                    System.out.println( "filter-- " +  o );
                }
            })   ;

      运行效果

     

    5、 消息数量过滤操作符的使用    

    • take :取前n个数据
    • takeLast:取后n个数据
    • first 只发送第一个数据
    • last 只发送最后一个数据
    • skip() 跳过前n个数据发送后面的数据
    • skipLast() 跳过最后n个数据,发送前面的数据
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    //take 发送前3个数据
           Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable.take( 3 )
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "take-- " +  o );
                       }
                   })   ;
     
           //takeLast 发送最后三个数据
           Observable observable2 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable2.takeLast( 3 )
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "takeLast-- " +  o );
                       }
                   })   ;
     
           //first 只发送第一个数据
           Observable observable3 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable3.first()
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "first-- " +  o );
                       }
                   })   ;
     
           //last 只发送最后一个数据
           Observable observable4 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable4.last()
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "last-- " +  o );
                       }
                   })   ;
     
           //skip() 跳过前2个数据发送后面的数据
           Observable observable5 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable5.skip( 2 )
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "skip-- " +  o );
                       }
                   })   ;
     
           //skipLast() 跳过最后两个数据,发送前面的数据
           Observable observable6 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
           observable5.skipLast( 2 )
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println( "skipLast-- " +  o );
                       }
                   })   ;
    效果图

     6、elementAt 、elementAtOrDefault

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    //elementAt() 发送数据序列中第n个数据 ,序列号从0开始
            //如果该序号大于数据序列中的最大序列号,则会抛出异常,程序崩溃
            //所以在用elementAt操作符的时候,要注意判断发送的数据序列号是否越界
     
            Observable observable7 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
            observable7.elementAt( 3 )
                    .subscribe(new Action1() {
                        @Override
                        public void call(Object o) {
                            System.out.println( "elementAt-- " +  o );
                        }
                    })   ;
     
            //elementAtOrDefault( int n , Object default ) 发送数据序列中第n个数据 ,序列号从0开始。
            //如果序列中没有该序列号,则发送默认值
            Observable observable9 = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;
            observable9.elementAtOrDefault(  8 , 666  )
                    .subscribe(new Action1() {
                        @Override
                        public void call(Object o) {
                            System.out.println( "elementAtOrDefault-- " +  o );
                        }
                    })   ;

      运行结果

     7、startWith() 插入数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    //插入普通数据
    //startWith 数据序列的开头插入一条指定的项 , 最多插入9条数据
    Observable observable = Observable.just( "aa" , "bb" , "cc" ) ;
    observable
            .startWith( "11" , "22" )
            .subscribe(new Action1() {
                @Override
                public void call(Object o) {
                    System.out.println( "startWith-- " + o );
                }
            }) ;
     
    //插入Observable对象
    List<String> list = new ArrayList<>() ;
    list.add( "ww" ) ;
    list.add( "tt" ) ;
    observable.startWith( Observable.from( list ))
            .subscribe(new Action1() {
                @Override
                public void call(Object o) {
                    System.out.println( "startWith2 -- " + o );
                }
            }) ;

      运行结果

     8、delay操作符,延迟数据发送

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    Observable<String> observable = Observable.just( "1" , "2" , "3" , "4" , "5" , "6" , "7" , "8" ) ;
     
           //延迟数据发射的时间,仅仅延时一次,也就是发射第一个数据前延时。发射后面的数据不延时
           observable.delay( 3 , TimeUnit.SECONDS )  //延迟3秒钟
                   .subscribe(new Action1() {
                       @Override
                       public void call(Object o) {
                           System.out.println("delay-- " + o);
                       }
                   }) ;

      

    9、Timer  延时操作符的使用

          使用场景:xx秒后,执行xx      

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    //5秒后输出 hello world , 然后显示一张图片
            Observable.timer( 5 , TimeUnit.SECONDS )
                    .observeOn(AndroidSchedulers.mainThread() )
                    .subscribe(new Action1<Long>() {
                        @Override
                        public void call(Long aLong) {
                            System.out.println( "timer--hello world " +  aLong );
                            findViewById( R.id.image).setVisibility(View.VISIBLE );
                        }
                    }) ;

         timer 返回一个 Observable , 它在延迟一段给定的时间后发射一个简单的数字0

        timer 操作符默认在computation调度器上执行,当然也可以用 Scheduler 在定义执行的线程。

     delay 、timer 总结: 

    •  相同点:delay 、 timer 都是延时操作符。
    •  不同点:delay  延时一次,延时完成后,可以连续发射多个数据。timer延时一次,延时完成后,只发射一次数据。

                                                 

     10、interval 轮询操作符,循环发送数据,数据从0开始递增

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    package app.com.myapplication;
    import android.support.v7.app.AppCompatActivity;
    import android.os.Bundle;
    import java.util.concurrent.TimeUnit;
    import rx.Observable;
    import rx.Subscription;
    import rx.functions.Action1;
     
    public class IntervalActivity extends AppCompatActivity {
     
        Subscription subscription ;
     
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_interval);
     
            //参数一:延迟时间  参数二:间隔时间  参数三:时间颗粒度
            Observable observable =  Observable.interval(3000, 3000, TimeUnit.MILLISECONDS) ;
            subscription = observable.subscribe(new Action1() {
                @Override
                public void call(Object o) {
                    System.out.println( "interval-  " + o );
                }
            })  ;
        }
     
        @Override
        protected void onDestroy() {
            super.onDestroy();
            if ( subscription != null ){
                subscription.unsubscribe();
            }
        }
    }

      

     11、doOnNext() 操作符,在每次 OnNext() 方法被调用前执行

           使用场景:从网络请求数据,在数据被展示前,缓存到本地

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    Observable observable = Observable.just( "1" , "2" , "3" , "4" ) ;<br>
           observable.doOnNext(new Action1() {
               @Override
               public void call(Object o) {
                   System.out.println( "doOnNext--缓存数据" + o  );
               }
           })
                   .subscribe(new Observer() {
                       @Override
                       public void onCompleted() {
     
                       }
     
                       @Override
                       public void onError(Throwable e) {
     
                       }
     
                       @Override
                       public void onNext(Object o) {
                           System.out.println( "onNext--" + o  );
                       }
                   }) ;

      

     12、Buffer 操作符

    • Buffer( int n )      把n个数据打成一个list包,然后再次发送。
    • Buffer( int n , int skip)   把n个数据打成一个list包,然后跳过第skip个数据。

             

        使用场景:一个按钮每点击3次,弹出一个toast      

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    List<String> list = new ArrayList<>();
          for (int i = 1; i < 10; i++) {
              list.add("" + i);
          }
     
          Observable<String> observable = Observable.from(list);
          observable
                  .buffer(2)   //把每两个数据为一组打成一个包,然后发送
                  .subscribe(new Action1<List<String>>() {
                      @Override
                      public void call(List<String> strings) {
                          System.out.println( "buffer---------------" );
                          Observable.from( strings ).subscribe(new Action1<String>() {
                              @Override
                              public void call(String s) {
                                  System.out.println( "buffer data --" + s  );
                              }
                          }) ;
                      }
                  });

      

         例子2: 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    //第1、2 个数据打成一个数据包,跳过第三个数据 ; 第4、5个数据打成一个包,跳过第6个数据
           observable.buffer( 2 , 3 //把每两个数据为一组打成一个包,然后发送。第三个数据跳过去
                   .subscribe(new Action1<List<String>>() {
                       @Override
                       public void call(List<String> strings) {
     
                           System.out.println( "buffer22---------------" );
                           Observable.from( strings ).subscribe(new Action1<String>() {
                               @Override
                               public void call(String s) {
                                   System.out.println( "buffer22 data --" + s  );
                               }
                           }) ;
                       }
                   }) ;

      

     13、throttleFirst 操作符

             在一段时间内,只取第一个事件,然后其他事件都丢弃。

             使用场景:1、button按钮防抖操作,防连续点击   2、百度关键词联想,在一段时间内只联想一次,防止频繁请求服务器   

    复制代码
     Observable.interval( 1 , TimeUnit.SECONDS)
                    .throttleFirst( 3 , TimeUnit.SECONDS )
                    .subscribe(new Action1<Long>() {
                        @Override
                        public void call(Long aLong) {
                            System.out.println( "throttleFirst--" + aLong );
                        }
                    }) ;
    复制代码

    这段代码,是循环发送数据,每秒发送一个。throttleFirst( 3 , TimeUnit.SECONDS )   在3秒内只取第一个事件,其他的事件丢弃。

    运行结果

    14、distinct    过滤重复的数据

    复制代码
     List<String> list = new ArrayList<>() ;
            list.add( "1" ) ;
            list.add( "2" ) ;
            list.add( "1" ) ;
            list.add( "3" ) ;
            list.add( "4" ) ;
            list.add( "2" ) ;
            list.add( "1" ) ;
            list.add( "1" ) ;
    
            Observable.from( list )
                    .distinct()
                    .subscribe(new Action1<String>() {
                        @Override
                        public void call(String s) {
                            System.out.println( "distinct--" + s );
                        }
                    }) ;
    复制代码

    从结果可以看出,重复的数据已经被过滤掉了

      distinctUntilChanged()  过滤连续重复的数据

    复制代码
    List<String> list = new ArrayList<>() ;
            list.add( "1" ) ;
            list.add( "2" ) ;
            list.add( "1" ) ;
            list.add( "3" ) ;
            list.add( "4" ) ;
            list.add( "4" ) ;
            list.add( "2" ) ;
            list.add( "1" ) ;
            list.add( "1" ) ;
    
            Observable.from( list )
                    .distinctUntilChanged()
                    .subscribe(new Action1<String>() {
                        @Override
                        public void call(String s) {
                            System.out.println( "distinctUntilChanged--" + s );
                        }
                    }) ;
    复制代码

     运行结果

    从结果可以看出,连续重复的数据已经被过滤掉了

    15、debounce() 操作符

           一段时间内没有变化,就会发送一个数据。

           使用场景:百度关键词联想提示。在输入的过程中是不会从服务器拉数据的。当输入结束后,在400毫秒没有输入就会去获取数据。

                       避免了,多次请求给服务器带来的压力.

     16、doOnSubscribe() 

            使用场景: 可以在事件发出之前做一些初始化的工作,比如弹出进度条等等

             注意:

                 1、doOnSubscribe() 默认运行在事件产生的线程里面,然而事件产生的线程一般都会运行在 io 线程里。那么这个时候做一些,更新UI的操作,是线程不安全的。

                     所以如果事件产生的线程是io线程,但是我们又要在doOnSubscribe() 更新UI , 这时候就需要线程切换。

                 2、如果在 doOnSubscribe() 之后有 subscribeOn() 的话,它将执行在离它最近的 subscribeOn() 所指定的线程。   

                 3、 subscribeOn() 事件产生的线程 ; observeOn() : 事件消费的线程

    复制代码
        Observable.create(onSubscribe)
        .subscribeOn(Schedulers.io())
        .doOnSubscribe(new Action0() {
            @Override
            public void call() {
                progressBar.setVisibility(View.VISIBLE); // 需要在主线程执行
            }
        })
        .subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(subscriber);
    复制代码

    17、range 操作符的使用 

        首先看range 方法的源码

     

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    public static Observable<Integer> range(int start, int count) {
         if (count < 0) {
             throw new IllegalArgumentException("Count can not be negative");
         }
         if (count == 0) {
             return Observable.empty();
         }
         if (start > Integer.MAX_VALUE - count + 1) {
             throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
         }
         if(count == 1) {
             return Observable.just(start);
         }
         return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
     }
     
      
     //可以通过第三个参数控制range执行的线程
     public static Observable<Integer> range(int start, int count, Scheduler scheduler) {
         return range(start, count).subscribeOn(scheduler);
     }

    Range操作符发射一个范围内的有序整数序列,你可以指定范围的起始和长度。

    RxJava将这个操作符实现为range函数,它接受两个参数,一个是范围的起始值,一个是范围的数据的数目。如果你将第二个参数设为0,将导致Observable不发射任何数据(如果设置为负数,会抛异常)。

    range默认不在任何特定的调度器上执行。有一个变体可以通过可选参数指定Scheduler。

    例子

    1
    2
    3
    4
    5
    6
    7
    Observable.range( 10 , 3 )
                  .subscribe(new Action1<Integer>() {
                      @Override
                      public void call(Integer integer) {
                          Log.v( "rx_range  " , "" + integer ) ;
                      }
                  }) ;

      结果

    /rx_range: 10
    /rx_range: 11
    /rx_range: 12

     18、defer 操作符

    例子

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    package app.com.myapplication;
     
    import android.support.v7.app.AppCompatActivity;
    import android.os.Bundle;
    import android.util.Log;
    import rx.Observable;
    import rx.functions.Action1;
    import rx.functions.Func0;
     
    public class DeferActivity extends AppCompatActivity {
     
        String i = "10" ;
     
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_defer);
     
            i = "11 " ;
     
            Observable<String> defer = Observable.defer(new Func0<Observable<String>>() {
                @Override
                public Observable<String> call() {
                    return Observable.just( i ) ;
                }
            }) ;
     
            Observable test = Observable.just(  i ) ;
     
            i = "12" ;
     
            defer.subscribe(new Action1<String>() {
                @Override
                public void call(String s) {
                    Log.v( "rx_defer  " , "" + s ) ;
                }
            }) ;
     
            test.subscribe(new Action1() {
                @Override
                public void call(Object o) {
                    Log.v( "rx_just " , "" + o ) ;
                }
            }) ;
        }
    }

      结果

    /rx_defer: 12
    /rx_just: 11

    • 可以看到,just操作符是在创建Observable就进行了赋值操作,而defer是在订阅者订阅时才创建Observable,此时才进行真正的赋值操作。
    • Defer操作符会一直等待直到有观察者订阅它,然后它使用Observable工厂方法生成一个Observable。它对每个观察者都这样做,因此尽管每个订阅者都以为自己订阅的是同一个Observable,事实上每个订阅者获取的是它们自己的单独的数据序列。
    • 在某些情况下,等待直到最后一分钟(就是知道订阅发生时)才生成Observable可以确保Observable包含最新的数据。

    RxJava 和 RxAndroid 三(生命周期控制和内存优化)

  • 相关阅读:
    ps入门
    ls命名 | Linux统计文件夹内的文件个数
    python打包成可执行文件
    装饰器
    poj 2752 Seek the Name, Seek the Fame
    字符串最大值
    次小生成树
    Selecting Courses
    poj 3461 Oulipo
    poj 2406 Power Strings
  • 原文地址:https://www.cnblogs.com/ldq2016/p/6594594.html
Copyright © 2011-2022 走看看