RxJava的车间操作工人-操作符

2019-11-09 14:04:49

字体：大中小

来源：转载

供稿：网友

在Rxjava中，如果把整个事件流看作是工厂的流水线，Observable就是原料，Observer就是我们的产品经理，这个产品是怎么交到我们产品经理手上的呢？中间很重要的就是工人，也就是操作符。它负责在Observable发出的事件和Observable的响应之间做一些处理。

首先我们来看一段Java代码：

static List<Student> studentList = new ArrayList<Student>(){ { add(new Student("Stay", 28)); add(new Student("谷歌小弟", 23)); add(new Student("Star", 25)); } };List<Student> list = new ArrayList<Student>(); new Thread(new Runnable(){ @Override public void run(){ synchronized (studentList) { for(Student student : studentList) { if(student.age > 23){ list.add(student); } } } runOnUiThread(new Runnable() { @Override public void run() { //UI显示过滤后的数据 displayUI(list); } }); } }).start();

此段代码实现的是过滤数据，也就是统计年龄大于23的学生，并将过滤数据显示在UI上。

使用RxJava来实现（体验一下RxJava的链式结构）：

Observable.from(studentList) //from可以理解为创建一个循环 .filter(new Func1<Student, Boolean>() {//filter代表根据规则过滤 @Override public Boolean call(Student student) { return student.age > 23; //过滤年龄大于23的学生 } }).subscribeOn(Schedulers.newThread()) //任务在新线程里运行 .toList() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1<List<Student>>() { @Override public void call(List<Student> list) { //UI显示过滤后的数据 displayUI(list); } });

可以看到，代码减少了，而且逻辑也很清晰。

下面我们再看个复杂点的场景：找出SD卡所有的.png格式的文件。

String basePath = Environment.getExternalStorageDirectory().getPath(); File rootFile = new File(basePath); Observable.just(rootFile) .flatMap(new Func1<File, Observable<File>>() { @Override public Observable<File> call(File file) {//遍历文件夹 return listFiles(file); } }) .filter(new Func1<File, Boolean>() {//过滤图片，二级文件夹中的图片无法找出 @Override public Boolean call(File file) { return file.getName().endsWith(".png"); } }) .map(new Func1<File, String>() { @Override public String call(File file) { return file.getPath(); } }) .toList() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<List<String>>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(List<String> list) {//返回png格式图片列表 //do something } });

其中，listFiles方法如下：

/** * 递归查询目录中的文件 * @param f * @return */ public static Observable<File> listFiles(final File f){ if(f.isDirectory()){ return Observable.from(f.listFiles()).flatMap(new Func1<File, Observable<File>>() { @Override public Observable<File> call(File file) { /**如果是文件夹就递归**/ return listFiles(file); } }); } else { /**filter操作符过滤,是文件就通知观察者**/ return Observable.just(f).filter(new Func1<File, Boolean>() { @Override public Boolean call(File file) { return f.exists() && f.canRead() ; } }); } }

从上面这段代码我们可以看到：虽然代码量看起来变复杂了，但是RxJava的实现是一条链式调用，没有任何的嵌套；整个实现逻辑看起来非常简洁清晰，这对我们的编程实现和后期维护是有巨大帮助的。特别是对于那些回调嵌套的场景。

有没有心动，喜欢上RxJava了呢？

将非Rx方法转换成Rx方法

/** * 模拟从数据库获取课程列表 * @return */PRivate static List<String> getCoursesFromDatabase() { List<String> courseList = new ArrayList<>(); courseList.add("美术"); courseList.add("体育"); courseList.add("音乐"); return courseList;}

在RxJava中，just和from操作符可以把一个对象转换成Observable，上面的方法使用just改写：

private static Observable<List<String>> getAllCourses() { return Observable.just(getCoursesFromDatabase());}

不过这样改写会有个问题，只要调用getAllCourses方法就会运行getCoursesFromDatabase方法，那么就不能通过指定Schedules的方式实现异步查询了。

所以如果某个方法经常会被异步调用，我们往往会使用create来改写：

private static Observable<List<String>> getAllCourses() { return Observable.create(new OnSubscribe<List<String>>() { @Override public void call(Subscriber<? super List<String>> subscriber) { List<String> courseList = getCoursesFromDatabase(); subscriber.onNext(courseList); subscriber.onCompleted(); } }); }

在使用Create操作符时，getCoursesFromDatabase方法是写在OnSubscribe接口的call方法里。调用getAllCourses方法，getCoursesFromDatabase并不会马上运行。

getAllCourses(db) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<String>(){...});

注意：因为just()，from()这类创建Observable的操作符在创建之初，就已经存储了对象的值，而不是在被订阅的时候才创建。所以在我们订阅之前，getCoursesFromDatabase()方法就已经在开始执行了，这样就不能达到我们想要的效果。