【Android】RxJava之初始篇

关于RxJava

RxJava是ReactiveX推出在Java VM环境下使用的异步操作库。除了在Java环境ReactiveX也为其他编程语言推出Rx库，例如Py、Js、Go等。网上有很多关于对RxJava的介绍和使用，在Android开发中也有很多项目使用RxJava。那为什么还要使用RxJava呢，Android开发也有提供异步操作的方法供开发者使用，我想应该是RxJava比起Handle、AsyncTask简洁优雅。

  • 1 RxJava采用链式调用，在程序逻辑上清晰简洁
  • 2 采用扩展式观察者设计模式

关于观察者模式以及其他RxJava的介绍这个就不做重复，下面内容主要围绕RxJava和RxAndroid使用。对于RxJava官方文档已经有详细介绍，本节是以学习讨论为主，如存在错误的地方希望大家可以指出。

被观察者Observable

使用RxJava需要创建Observable，Observable用于发射数据。如下Observable的create方法需要传入一个OnSubscribe，其继承于Action1>，Action中的Subscriber就是订阅者。

 
 
 
 
    
  
  
  
  
  1. public static  Observable create(OnSubscribe f) { 
    2. 
  
  
  
  
  2.     return new Observable(RxJavaHooks.onCreate(f)); 
    3. 
  
  
  
  
  3. }  
    4.

另外create方法中需要实现接口call，返回subscriber对象。call方法实现在observable订阅后要执行的事件流。subscriber.onNext发射data，subscriber.onCompleted可以表示发射事件结束。接着调用observable的subscribe方法实现被订阅后执行的事件流。 

 
 
 
 
    
  
  
  
  
  1. Observable observable = Observable 
    2. 
  
  
  
  
  2.                 .create(new Observable.OnSubscribe() { 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.             @Override 
    5. 
  
  
  
  
  5.             public void call(Subscriber subscriber) { 
    6. 
  
  
  
  
  6.                 subscriber.onNext("1"); 
    7. 
  
  
  
  
  7.                 subscriber.onNext("2"); 
    8. 
  
  
  
  
  8.                 subscriber.onNext("3"); 
    9. 
  
  
  
  
  9.                 subscriber.onNext("4"); 
    10. 
  
  
  
  
  10.                 subscriber.onNext("5"); 
    11. 
  
  
  
  
  11.             } 
    12. 
  
  
  
  
  12. }); 
    13. 
  
  
  
  
  13. Subscriber subscriber = new  Subscriber() { 
    14. 
  
  
  
  
  14.             @Override 
    15. 
  
  
  
  
  15.             public void onCompleted() { 
    16. 
  
  
  
  
  16.  
    17. 
  
  
  
  
  17.             } 
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19.             @Override 
    20. 
  
  
  
  
  20.             public void onError(Throwable e) { 
    21. 
  
  
  
  
  21.  
    22. 
  
  
  
  
  22.             } 
    23. 
  
  
  
  
  23.  
    24. 
  
  
  
  
  24.             @Override 
    25. 
  
  
  
  
  25.             public void onNext(String s) { 
    26. 
  
  
  
  
  26.                 System.out.print(s + '\n'); 
    27. 
  
  
  
  
  27.             } 
    28. 
  
  
  
  
  28. }; 
    29. 
  
  
  
  
  29. observable.subscribe(subscriber); 
    30. 
  
  
  
  
  30. //输出结果 print： 
    31. 
  
  
  
  
  31. //1 
    32. 
  
  
  
  
  32. //2 
    33. 
  
  
  
  
  33. //3 
    34. 
  
  
  
  
  34. //4 
    35. 
  
  
  
  
  35. //5   
    36.

Observable除了使用create方法创建外还可以使用from或者just快速设置发射的事件流，简化了create的步骤。 

  

 
 
 
 
    
  
  
  
  
  1. Observable o = Observable.from("a", "b", "c");  
    2.

 

  

 
 
 
 
    
  
  
  
  
  1. Observable o = Observable.just("one object"); 
    2.

说好的异步操作

RxJava的线程由Schedulers调度者控制，通过它来控制具体操作在什么线程中进行。

  • Schedulers.immediate() 在当前线程中执行
  • Schedulers.newThread() 为每一个任务开辟线程执行
  • Schedulers.computation() 计算任务运行的线程
  • Schedulers.io() IO任务运行的线程....
  • AndroidSchedulers.mainThread() Android 主线程运行

对于线程的控制主要由subscribeOn()和observeOn()两个方法控制：

  • subscribeOn 控制Observable.OnSubscribe所处的线程，等同于Observable create、just、from时所处的线程。
  • observeOn 控制Subscriber的线程，也可以说是控制事件被执行时所在的线程。
 
 
 
 
    
  
  
  
  
  1. Observable 
    2. 
  
  
  
  
  2.         .just(1,2,3) 
    3. 
  
  
  
  
  3.         .subscribeOn(Schedulers.io()) 
    4. 
  
  
  
  
  4.         .observeOn(AndroidSchedulers.mainThread()) 
    5. 
  
  
  
  
  5.         .subscribe(new Subscriber() { 
    6. 
  
  
  
  
  6.             @Override 
    7. 
  
  
  
  
  7.             public void onCompleted() { 
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9.             } 
    10. 
  
  
  
  
  10.  
    11. 
  
  
  
  
  11.             @Override 
    12. 
  
  
  
  
  12.             public void onError(Throwable e) { 
    13. 
  
  
  
  
  13.  
    14. 
  
  
  
  
  14.             } 
    15. 
  
  
  
  
  15.  
    16. 
  
  
  
  
  16.             @Override 
    17. 
  
  
  
  
  17.             public void onNext(Integer integer) { 
    18. 
  
  
  
  
  18.                  System.out.print(integer + '\n');                        
    19. 
  
  
  
  
  19.             } 
    20. 
  
  
  
  
  20. }); 
    21. 
  
  
  
  
  21. //输出结果 print： 
    22. 
  
  
  
  
  22. //1 
    23. 
  
  
  
  
  23. //2 
    24. 
  
  
  
  
  24. //3  
    25.

写下上面的RxJava链式调用的代码，有没有觉得比以前使用的异步调用清爽许多，对处女座还说这很治愈!

操作符Operators

ReactiveX提供超级多的操作符，每个操作符都具有不同的功能，但目的都是在Observable和Subscribe之间变换和修改发射出去的事件流。这节介绍几个比较常见简单的操作符，之后有机会再写一节操作符篇详细说说每个操作符的作用。附上官方操作符文档看看就知道有多少多了。

  • Map() 
 
 
 
 
    
  
  
  
  
  1. public final  Observable map(Func1 func) { 
    2. 
  
  
  
  
  2.         return create(new OnSubscribeMap(this, func)); 
    3. 
  
  
  
  
  3.     }   
    4.

 

首先先介绍一个操作符map，map实现Func1接口将T类型数据变换为R类型数据，返回R类型数据。例如传入Integer类型的事件队列，经过map加工之后以String类型返回。 

 
 
 
 
    
  
  
  
  
  1. Observable 
    2. 
  
  
  
  
  2.                 .just(1,2,3) 
    3. 
  
  
  
  
  3.                 .subscribeOn(Schedulers.io()) 
    4. 
  
  
  
  
  4.                 .observeOn(AndroidSchedulers.mainThread()) 
    5. 
  
  
  
  
  5.                 .map(new Func1() { 
    6. 
  
  
  
  
  6.                     @Override 
    7. 
  
  
  
  
  7.                     public String call(Integer integer) { 
    8. 
  
  
  
  
  8.                         return integer + ""; 
    9. 
  
  
  
  
  9.                     } 
    10. 
  
  
  
  
  10.                 }) 
    11. 
  
  
  
  
  11.                 .subscribe(new Subscriber() { 
    12. 
  
  
  
  
  12.                     ...... 
    13. 
  
  
  
  
  13.                     @Override 
    14. 
  
  
  
  
  14.                     public void onNext(String str) { 
    15. 
  
  
  
  
  15.                         System.out.print(str + '\n'); 
    16. 
  
  
  
  
  16.                     } 
    17. 
  
  
  
  
  17.                 }); 
    18. 
  
  
  
  
  18. //输出结果 print： 
    19. 
  
  
  
  
  19. //1 
    20. 
  
  
  
  
  20. //2 
    21. 
  
  
  
  
  21. //3  
    22.
  • Filter() 
 
 
 
 
    
  
  
  
  
  1. public final Observable filter(Func1 predicate) { 
    2. 
  
  
  
  
  2.        return create(new OnSubscribeFilter(this, predicate)); 
    3. 
  
  
  
  
  3.    }   
    4.

 

filter和map一样实现Func1接口不过它变换之后的类型为boolean，对发射的事件流进行筛选，当变换后的boolean值为true，订阅者才能收到通过筛选的事件，反之该事件不被消费。例如事件流筛选要求当int值可被2整除才能继续传递，所以最后订阅者可消费的事件为2,4,6,8,10。

 
 
 
 
    
  
  
  
  
  1. Observable 
    2. 
  
  
  
  
  2.                 .just(1,2,3,4,5,6,7,8,9,10) 
    3. 
  
  
  
  
  3.                 .subscribeOn(Schedulers.io()) 
    4. 
  
  
  
  
  4.                 .observeOn(AndroidSchedulers.mainThread()) 
    5. 
  
  
  
  
  5.                 .filter(new Func1() { 
    6. 
  
  
  
  
  6.                     @Override 
    7. 
  
  
  
  
  7.                     public Boolean call(Integer integer) { 
    8. 
  
  
  
  
  8.                         return integer % 2 == 0; 
    9. 
  
  
  
  
  9.                     } 
    10. 
  
  
  
  
  10.                 }) 
    11. 
  
  
  
  
  11.                 .map(new Func1() { 
    12. 
  
  
  
  
  12.                     @Override 
    13. 
  
  
  
  
  13.                     public String call(Integer integer) { 
    14. 
  
  
  
  
  14.                         return integer + ""; 
    15. 
  
  
  
  
  15.                     } 
    16. 
  
  
  
  
  16.                 }) 
    17. 
  
  
  
  
  17.                 .subscribe(new Subscriber() { 
    18. 
  
  
  
  
  18.                    ...... 
    19. 
  
  
  
  
  19.                     @Override 
    20. 
  
  
  
  
  20.                     public void onNext(String str) { 
    21. 
  
  
  
  
  21.                         System.out.print(str + '\n'); 
    22. 
  
  
  
  
  22.                         Log.i("subscribe", str); 
    23. 
  
  
  
  
  23.                     } 
    24. 
  
  
  
  
  24.                 }); 
    25. 
  
  
  
  
  25. //输出结果 print： 
    26. 
  
  
  
  
  26. //2 
    27. 
  
  
  
  
  27. //3 
    28. 
  
  
  
  
  28. //4 
    29. 
  
  
  
  
  29. //6 
    30. 
  
  
  
  
  30. //8 
    31. 
  
  
  
  
  31. //10  
    32.
  • Skip()
 
 
 
 
    
  
  
  
  
  1. public final Observable skip(int count) { 
    2. 
  
  
  
  
  2.         return lift(new OperatorSkip(count)); 
    3. 
  
  
  
  
  3.     }  
    4.

 

skip操作符表示跳过前几个事件从某一个事件开始发射事件，下标从0开始。

 
 
 
 
    
  
  
  
  
  1.  Observable 
    2. 
  
  
  
  
  2.                 .just(1,2,3,4,5,6,7,8,9,10) 
    3. 
  
  
  
  
  3.                 .subscribeOn(Schedulers.io()) 
    4. 
  
  
  
  
  4.                 .observeOn(AndroidSchedulers.mainThread()) 
    5. 
  
  
  
  
  5.                 .skip(3) 
    6. 
  
  
  
  
  6.                 .map(new Func1() { 
    7. 
  
  
  
  
  7.                     @Override 
    8. 
  
  
  
  
  8.                     public String call(Integer integer) { 
    9. 
  
  
  
  
  9.                         return integer + ""; 
    10. 
  
  
  
  
  10.                     } 
    11. 
  
  
  
  
  11.                 }) 
    12. 
  
  
  
  
  12.                 .subscribe(new Subscriber() { 
    13. 
  
  
  
  
  13.                     ...... 
    14. 
  
  
  
  
  14.                     @Override 
    15. 
  
  
  
  
  15.                     public void onNext(String s) { 
    16. 
  
  
  
  
  16.                         System.out.print(s + '\n'); 
    17. 
  
  
  
  
  17.                         Log.i("subscribe", s); 
    18. 
  
  
  
  
  18.                     } 
    19. 
  
  
  
  
  19.                 }); 
    20. 
  
  
  
  
  20. //输出结果 print： 
    21. 
  
  
  
  
  21. //4 
    22. 
  
  
  
  
  22. //5 
    23. 
  
  
  
  
  23. //6 
    24. 
  
  
  
  
  24. //7 
    25. 
  
  
  
  
  25. //8 
    26. 
  
  
  
  
  26. //9 
    27. 
  
  
  
  
  27. //10  
    28.
  • Range()
 
 
 
 
    
  
  
  
  
  1. public static Observable range(int start, int count) { 
    2. 
  
  
  
  
  2.         if (count < 0) { 
    3. 
  
  
  
  
  3.             throw new IllegalArgumentException("Count can not be negative"); 
    4. 
  
  
  
  
  4.         } 
    5. 
  
  
  
  
  5.         if (count == 0) { 
    6. 
  
  
  
  
  6.             return Observable.empty(); 
    7. 
  
  
  
  
  7.         } 
    8. 
  
  
  
  
  8.         if (start > Integer.MAX_VALUE - count + 1) { 
    9. 
  
  
  
  
  9.             throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE"); 
    10. 
  
  
  
  
  10.         } 
    11. 
  
  
  
  
  11.         if(count == 1) { 
    12. 
  
  
  
  
  12.             return Observable.just(start); 
    13. 
  
  
  
  
  13.         } 
    14. 
  
  
  
  
  14.         return Observable.create(new OnSubscribeRange(start, start + (count - 1))); 
    15. 
  
  
  
  
  15.     }   
    16.

 

range操作符可以理解为just，from传递一个连续的int类型待发射数组，n为起始int值，m为Count。例如n = 1，m = 5 int数组就是{1，2，3，4，5}

结尾

这节先学习到这，算是对RxJava的初步认识和学习。其实运用RxJava主要还是依赖于对操作符的使用，前面所介绍的操作符属于最最简单基础的，还有很多特别有用的操作符没有介绍。之后再继续介绍一些操作符。RxJava在Android开发中很受欢迎，就是因为它的强大，同时RxJava可以和Retrofit组合使用，更高效处理网络请求回值。另外GitHub GoogleSample上的android-architecture也有使用RxJava框架的TODO项目，可以看看理解RxJava在项目中的实践应用。

当前文章：【Android】RxJava之初始篇
分享URL：http://www.shufengxianlan.com/qtweb/news35/469785.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

外贸建站知识

同城分类信息

梓潼做网站　　 　网络推广外包　　 　成都主机托管　　 　成都定制网站建设　　 　手机网站建设　　 　成都防护栏　　 　成都外贸网站优化　　 　雅安托管服务器　　 　彭州做网站　　 　香港服务器空间　　 　泸州网站建设　　 　子晨包装　　 　成都娃娃鱼　　 　四川服务器托管　　 　娄底尾呼妄服装　　 　成都托管服务器　　 　网站推广公司　　 　重庆企业网站建设　　 　自动门　　 　温江做网站