RxJava使用场景总结

冷不防 2022-02-24 15:12 284阅读 0赞

RxJava早已经成为java圈最近3年来最流行的类库，无数优秀的开源框架中对其进行接入。但什么场景下使用，其实没有定论，以下是个人整理总结的。

### 1. 代码迷之缩进 ###

代码中存在大量的缩进，对其可读性非常差，而RxJava提供了非常多的函数足以胜任你98%的需求。

//这么多缩进，难受不
    public void getPng() {
        for (File folder : folders) {
            File[] files = folder.listFiles();
            for (File file : files) {
                if (file.getName().endsWith(".png")) {
                    final Bitmap bitmap = getBitmapFromFile(file);
                    getActivity().runOnUiThread(new Runnable() {
                        @Override
                        public void run() {
                            imageCollectorView.addImage(bitmap);
                        }
                    });
                }
            }
        }
    }
    //也可以很漂亮
    Observable.from(folders)
    .flatMap(new Func1<File, Observable<File>>() {
        @Override
        public Observable<File> call(File file) {
            return Observable.from(file.listFiles());
        }
    })
    .filter(new Func1<File, Boolean>() {
        @Override
        public Boolean call(File file) {
            return file.getName().endsWith(".png");
        }
    })
    .map(new Func1<File, Bitmap>() {
        @Override
        public Bitmap call(File file) {
            return getBitmapFromFile(file);
        }
    })
    .subscribe(new Action1<Bitmap>() {
        @Override
        public void call(Bitmap bitmap) {
            imageCollectorView.addImage(bitmap);
        }
    });
    //或者采用这样
    //有Action0、Action1...到Action9
    Action1<String> onNextAction = new Action1<String>() {
        @Override
        public void call(String s) {
            System.out.println(s);
        }
    };
    Action1<Throwable> onErrorAction = new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            //错误处理
        }
    };
    Action0 onCompleteAction = new Action0() {
        @Override
        public void call() {
            System.out.println("completed");
        }
    };
    Observable observable = Observable.just("hello");
    observable.subscribe(onNextAction, onErrorAction, onCompleteAction);

### 2. 切换线程/同步/异步 ###

这个就不讲了，本来就是RxJava擅长的事情

> subscribeOn(Schedulers.io())      //指定OnSubscribe被激活时处在的线程，事件产生线程  
> observeOn(Schedulers.single())  //Subscriber所运行的线程，事件消费的线程

### **3. RxJava代替EventBus进行数据传递** ###

[RxBus][]并不是一个库，而是一种模式，是使用了RxJava的思想来达到EventBus的数据传递效果。

### **4. 定时操作：timer** ###

当有“x秒后执行y操作”类似的需求的时候，但它不会阻塞当前线程。

//2秒后输出日志“hello world”，然后结束
    Observable.timer(2, TimeUnit.SECONDS)
            .subscribe(new Observer<Long>() {
                @Override
                public void onCompleted() {
                    System.out.println("completed");
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println("error");
                }
                @Override
                public void onNext(Long number) {
                    System.out.println("hello");
                }
            });

### 5. **周期性操作：interval** ###

当有“每隔xx秒后执行yy操作”类似的需求的时候

//每隔2秒输出日志“hello”，然后结束
    Observable.interval(2, TimeUnit.SECONDS)
            .subscribe(new Observer<Long>() {
                @Override
                public void onCompleted() {
                    System.out.println("completed");
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println("error");
                }
                @Override
                public void onNext(Long number) {
                    System.out.println("hello");
                }
            });

### 6. **合并两个数据源：merge** ###

例如一组数据来自网络，一组数据来自文件，需要合并两组数据一起展示。

@Test
    public void merge() {
        Observable.merge(getDataFromFile(), getDataFromNet())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("done loading all data");
                    }
                    @Override
                    public void onError(Throwable e) {
                        System.out.println("error");
                    }
                    @Override
                    public void onNext(String s) {
                        System.out.println("merge:" + s);
                    }
                });
    }
    
    private Observable<String> getDataFromFile() {
        String[] strs = {"filedata1", "filedata2", "filedata3", "filedata4"};
        //Thread.sleep(1000);
        Observable<String> temp = Observable.from(strs);
        return temp;
    }
    
    private Observable<String> getDataFromNet() {
        String[] strs = {"netdata1", "netdata2", "netdata3", "netdata4"};
        Observable<String> temp = Observable.from(strs);
        return temp;
    }

### 7. **使用concat和first做判断检查** ###

依次检查memory、disk和network中是否存在数据，任何一步一旦发现数据（*有执行onNext*）后面的操作都不执行。

String memoryCache = "x";
    Observable<String> memory = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            if (memoryCache != null) {
                subscriber.onNext(memoryCache);
            } else {
                subscriber.onCompleted();
            }
        }
    });
    String diskData = "1";
    Observable<String> disk = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            if ("1".equals(diskData)) {
                subscriber.onNext(diskData);
            } else {
                subscriber.onCompleted();
            }
        }
    });
    Observable<String> network = Observable.just("network");
    //依次检查memory、disk、network
    Observable.concat(memory, disk, network).first().subscribe(s -> System.out.println("--------------subscribe: " + s));

### 8. **防止按钮重复点击：throttleFirst和**debounce ###

一定时间内取发送的事件，每次事件发射就会重新计时，debounce也能达到同样的效果。二者区别在于：debounce第一次事件会被忽略响应处理。

Observable<Integer> tObservable = Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            try {
                subscriber.onNext(1);
                Thread.sleep(400);
                subscriber.onNext(2);
                Thread.sleep(505);
                subscriber.onNext(3);
                Thread.sleep(100);
                subscriber.onNext(4);
                Thread.sleep(450);
                subscriber.onNext(5);
                Thread.sleep(510);
                subscriber.onCompleted();
            } catch (InterruptedException e) {
                subscriber.onError(e);
            }
        }
    });
    //输出2,5
    tObservable.debounce(500, TimeUnit.MILLISECONDS)// 设置时间为0.5秒
            .subscribe(x -> System.out.println(x));
    System.out.println("throttleFirst:");
    //输出1,3,5
    tObservable.throttleFirst(500, TimeUnit.MILLISECONDS)// 设置时间为0.5秒
            .subscribe(x -> System.out.println(x));

### **9. 轮询请求：schedulePeriodically** ###

轮询请求，在一些需要反复需要获取数据的场景，比如过几十分钟更新一下天气数据之类的

Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(final Subscriber<? super String> subscriber) {
            Schedulers.io().createWorker()  //指定在io线程执行
                    .schedulePeriodically(new Action0() {
                        @Override
                        public void call() {
                            subscriber.onNext("doNetworkCallAndGetStringResult");
                        }
                    }, 2000, 1000, TimeUnit.MILLISECONDS);//初始延迟，polling延迟
        }
    }).subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            System.out.println("polling..." + s);
        }
    });

使用过**Rxjava**的小伙伴都知道，在使用RxJava时如果处理不当，很可能会产生**内存泄漏**的问题。为此便引进了CompositeSubscription类来统一管理。![89553999][]

[RxBus]: https://github.com/AndroidKnife/RxBus
[89553999]: /images/20220224/ec25265f0723468a926215d9a748031d.png