RxJava 教程-1 简介 原理 线程控制 变换

时间：2022-04-08 20:22:29

简介
RxJava 是什么？




RxJava 在 GitHub 主页上的自我介绍是






RxJava is a Java VM implementation of ReactiveX: a library for composing asynchronous and event-based programs by using observable sequences.




RxJava是 ReactiveX 在JVM上的一个实现：一个使用可观测的序列(observable sequences)来组成(composing )异步的(asynchronous )、基于事件(event-based)的程序的库。


其实， RxJava 的本质可以压缩为异步这一个词。说到根上，它就是一个实现异步操作的库，而别的定语都是基于这之上的。








RxJava 好在哪？


一个词：简洁。




异步操作很关键的一点是程序的简洁性，因为在调度过程比较复杂的情况下，异步代码经常会既难写也难被读懂。 Android 创造的AsyncTask 和Handler ，其实都是为了让异步代码更加简洁。RxJava 的优势也是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁。



RxJava 教程-1 简介 原理 线程控制 变换




假设有这样一个需求：界面上有一个自定义的视图，它的作用是显示多张图片，并能任意增加显示的图片。现在需要将一个给出的目录数组中每个目录下的 png 图片都加载出来并显示在View中。需要注意的是，由于读取图片的这一过程较为耗时，需要放在后台执行，而图片的显示则必须在 UI 线程执行。其中一种方式：






    

  1. new Thread() {
    2. 

  2.     @Override
    3. 

  3.     public void run() {
    4. 

  4.         super.run();
    5. 

  5.         for (File folder : folders) {
    6. 

  6.             File[] files = folder.listFiles();
    7. 

  7.             for (File file : files) {
    8. 

  8.                 if (file.getName().endsWith(".png")) {
    9. 

  9.                     final Bitmap bitmap = getBitmapFromFile(file);
    10. 

  10.                     runOnUiThread(new Runnable() {
    11. 

  11.                         @Override
    12. 

  12.                         public void run() {
    13. 

  13.                             imageCollectorView.addImage(bitmap);
    14. 

  14.                         }
    15. 

  15.                     });
    16. 

  16.                 }
    17. 

  17.             }
    18. 

  18.         }
    19. 

  19.     }
    20. 

  20. }.start();
    21. 







而如果使用 RxJava ，实现方式是这样的：






    

  1. Observable.from(folders)
    2. 

  2.     .flatMap(new Func1<File, Observable<File>>() {
    3. 

  3.         @Override
    4. 

  4.         public Observable<File> call(File file) {
    5. 

  5.             return Observable.from(file.listFiles());
    6. 

  6.         }
    7. 

  7.     })
    8. 

  8.     .filter(new Func1<File, Boolean>() {
    9. 

  9.         @Override
    10. 

  10.         public Boolean call(File file) {
    11. 

  11.             return file.getName().endsWith(".png");
    12. 

  12.         }
    13. 

  13.     })
    14. 

  14.     .map(new Func1<File, Bitmap>() {
    15. 

  15.         @Override
    16. 

  16.         public Bitmap call(File file) {
    17. 

  17.             return getBitmapFromFile(file);
    18. 

  18.         }
    19. 

  19.     })
    20. 

  20.     .subscribeOn(Schedulers.io())
    21. 

  21.     .observeOn(AndroidSchedulers.mainThread())
    22. 

  22.     .subscribe(new Action1<Bitmap>() {
    23. 

  23.         @Override
    24. 

  24.         public void call(Bitmap bitmap) {
    25. 

  25.             imageCollectorView.addImage(bitmap);
    26. 

  26.         }
    27. 

  27.     });
    28. 







观察一下你会发现， RxJava 的这个实现，是一条从上到下的链式调用，没有任何【嵌套】，这在逻辑的简洁性上是具有优势的。当需求变得复杂时，这种优势将更加明显。


另外，如果你的 IDE 是 Android Studio ，其实每次打开某个 Java 文件的时候，你会看到被自动 Lambda 化的预览，这将让你更加清晰地看到程序逻辑。






    

  1. Observable.from(folders)
    2. 

  2.     .flatMap((Func1) (folder) -> { Observable.from(file.listFiles()) })
    3. 

  3.     .filter((Func1) (file) -> { file.getName().endsWith(".png") })
    4. 

  4.     .map((Func1) (file) -> { getBitmapFromFile(file) })
    5. 

  5.     .subscribeOn(Schedulers.io())
    6. 

  6.     .observeOn(AndroidSchedulers.mainThread())
    7. 

  7.     .subscribe((Action1) (bitmap) -> { imageCollectorView.addImage(bitmap) });
    8. 





如果你习惯使用 Retrolambda ，你也可以直接把代码写成上面这种简洁的形式。Retrolambda 是 Java 6/7 对 Lambda 表达式的非官方兼容方案，它的向后兼容性和稳定性是无法保障的，因此对于企业项目，使用 Retrolambda 是有风险的。











在Flipboard 的 Android 代码中，有一段逻辑非常复杂，包含了多次内存操作、本地文件操作和网络操作，对象分分合合，线程间相互配合相互等待，一会儿排成人字，一会儿排成一字。如果使用常规的方法来实现，肯定是要写得欲仙欲死，然而在使用 RxJava 的情况下，依然只是一条链式调用就完成了。它很长，但很清晰。


所以， RxJava 好在哪？就好在简洁，好在那把什么复杂逻辑都能穿成一条线的简洁。

原理


RxJava 的异步实现，是通过一种【扩展的观察者模式】来实现的。






RxJava 有四个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。


Observable 和Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。




与传统观察者模式不同， RxJava 的事件回调方法除了普通事件 onNext()之外，还定义了两个特殊的事件：onCompleted() 和 onError()。




    

  • onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的onNext() 发出时，需要触发 onCompleted() 方法作为标志。
    • 

  • onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
    • 





在一个正确运行的事件序列中, onCompleted() 和 onError() 


onCompleted() 和 onError() 二者是互斥的，有且只有一个，并且是事件序列中的最后一个。


基于以上的概念， RxJava 的基本实现主要有三点：
















1、创建 观察者Observer，它决定事件触发的时候将有怎样的行为。 


RxJava 中的 Observer 接口的实现方式：








    

  1. Observer<String> observer = new Observer<String>(){
    2. 

  2. 	
    3. 

  3. 	@Override
    4. 

  4. 
	public void onCompleted() {, 2, 3, 4).subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
    5. 

  5. 		.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
    6. 

  6. 		.subscribe(number -> {Log.i("bqt", "number:" + number); });
    7. 









事实上，这种在 subscribe() 之前写上两句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常见，它适用于多数的 『后台线程取数据，主线程显示』的程序策略。

变换


终于要到牛逼的地方了，不管你激动不激动，反正我是激动了。







RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。


所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。
















1、map()


首先看一个 map() 的例子。








    

  1. public class Bean {
    2. 

  2. 	public String name;
    3. 

  3. 	public int age;
    4. 

  4. 	public List<String> courses = Arrays.asList("语文", "数学", "英语");
    5. 

  5. 	
    6. 

  6. 	public Bean() {
    7. 

  7. 	}
    8. 

  8. 	
    9. 

  9. 	public Bean(String name, int age) {
    10. 

  10. 		this.name = name;
    11. 

  11. 		this.age = age;
    12. 

  12. 	}
    13. 

  13. 	
    14. 

  14. 	@Override
    15. 

  15. 	public String toString() {
    16. 

  16. 		return "name=" + name + "     age=" + age;
    17. 

  17. 	}
    18. 

  18. }
    19. 





打印一组学生的名字








    

  1. //打印出一组学生的名字
    2. 

  2. Bean[] beans = {new Bean("包青天", 27), new Bean("白乾涛", 26)};
    3. 

  3. Observable.from(beans)//
    4. 

  4. 		.map(new Func1<Bean, String>() {
    5. 

  5. 			@Override
    6. 

  6. 
			public String call(Bean bean) { //), new Bean("白乾涛", 26)};
    7. 

  7. Observable.from(beans)//
    8. 

  8. 		.map(bean -> {return bean.name;})// 输入类型为 bean，返回类型为String
    9. 

  9. 		.subscribe(name -> {Log.i("bqt", name);});
    10. 







可以看到，map() 方法将参数中的Bean 对象转换成一个 String 对象后返回，而在经过 map() 方法后，事件的参数类型也由 Bean转为了 String 。


这种直接变换对象并返回的，是最常见的也最容易理解的变换。不过 RxJava 的变换远不止这样，它不仅可以针对事件对象，还可以针对整个事件队列，这使得 RxJava 变得非常灵活。


map() 示意图：





RxJava 教程-1 简介 原理 线程控制 变换
















2、flatMap()


这是一个很有用但非常难理解的变换。


还是上面的需求，如果我要打印出每个学生所需要修的所有课程的名称（每个学生只有一个名字，但却有多个课程），该怎么做呢？




按照以前的思维，我们可以这样：






    

  1. Bean[] beans = {new Bean("包青天", 27), new Bean("白乾涛", 26)};
    2. 

  2. Observable.from(beans)//
    3. 

  3. 		.subscribe(bean -> {
    4. 

  4. 			List<String> courses = bean.courses;
    5. 

  5. 			for (int i = 0 ; i < courses.size() ; i++) {
    6. 

  6. 				Log.i("bqt", courses.get(i));
    7. 

  7. 			}
    8. 

  8. 		});
    9. 





依然很简单。那么如果我不想在 Subscriber 中使用 for 循环，而是希望 Subscriber 中直接传入单个的课程（String）对象呢（这对于代码复用很重要）？


用 map() 显然是不行的，因为 map() 是一对一的转化，而我现在的要求是一对多的转化。那怎么才能把一个 Student 转化成多个 String呢？




这个时候，就需要用 flatMap() 了：








    

  1. Bean[] beans = {new Bean("包青天", 27), new Bean("白乾涛", 26)};
    2. 

  2. Observable.from(beans)//
    3. 

  3. 		.flatMap(new Func1<Bean, Observable<String>>() {
    4. 

  4. 			@Override
    5. 

  5. 			public Observable<String> call(Bean bean) {
    6. 

  6. 				return Observable.from(bean.courses);
    7. 

  7. 			}
    8. 

  8. 		})
    9. 

  9. 		.subscribe(name -> {Log.i("bqt", name);});
    10. 







flatMap() 和 map() 有一个相同点：它也是把传入的参数转化之后返回另一个对象。但需要注意，和map() 不同的是， flatMap() 中返回的是一个 Observable 对象，但是这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中（意思是说，subscribe方法接受的参数不是创建的这个Observable 对象，而是这个Observable 对象发送的事件）。




flatMap() 的原理是这样的：




    

  • 1. 使用传入的事件对象创建一个 Observable 对象；
    • 

  • 2. 并不发送这个 Observable，而是将它激活，于是它开始发送事件；
    • 

  • 3. 每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable，而这个 Observable 负责将这些事件统一交给Subscriber 的回调方法。
    • 





这三个步骤，把事件拆成了两级，通过一组新创建的 Observable 将初始的对象『铺平』之后通过统一路径分发了下去。而这个『铺平』就是 flatMap() 所谓的 flat。


flatMap() 示意图：





RxJava 教程-1 简介 原理 线程控制 变换




包青天解释：




    

  • 1、最开始的Observable在被订阅后开始发送事件，此时发送的事件为两个圆形（比如上例中的Bean）
    • 

  • 2、在被flatMap后，每个圆形都生成了一个对应的Observable 对象，但是并不是直接把这两个Observable发送到Observer中
    • 

  • 3、而是将这两个Observable对象激活，于是这两个Observable 对象就开始发送事件了
    • 

  • 4、这两个Observable 对象各自发送了两个事件，此时发送的事件类型都为方形（比如上例中的String）
    • 

  • 5、这两个Observable 对象一共发送了四个事件，这四个事件又都被汇入到同一个Observable中（就像图中的那块云）
    • 

  • 6、最后由这个Observable对象负责将这四个事件发送给Observer中
    •

相关文章