RxJava入门

参考文档

https://www.jianshu.com/p/bb0499e8394e 

包含：

Android开发-RxJava详解.pdf（偏向于Android开发的指导）

RxJava Essentials_中文翻译版.pdf（国外大牛Ivan.Morgillo编写的RxJava教程）

RxJava官方中文文档.pdf（比较全面，包含所有操作符的说明。这是RxJava1的文档，目前通常使用RxJava2，RxJava2基于RxJava1扩充了操作符和少量修改，整体用法和思维是一致的）

1 RxJava基本介绍

1.1 概念

        RxJava是采用响应式编程思想的异步操作框架，设计初衷是让异步编程变得简洁，灵活。Android 提供的 AsyncTask 和Handler ，其实都是为了让异步代码更加简洁（逻辑简洁）。 RxJava 的优势也是简洁，但它的简洁的与众不同之处在于，随着程序逻辑变得越来越复杂，它依然能够保持简洁（逻辑简洁），这得益于RxJava编码特征：减少嵌套、多采用链式调用，简化代码逻辑。

1.2 内容概括

        两个对象 操作符

两个对象：被观察者、观察者

操作符（方法）：

• 创建操作：create、defer、empty/never/throw、from、interval、just、range、repeat、start、timer
• 变换操作：buffer、flatMap、groupBy、map、scan、window
• 过滤操作：debounce、distinct、elementAt、filter、first、ignoreElements、Last、Sample、Skip、SkipLast、Take、TakeLast
• 结合操作：and/then/when、combineLatest、join、merge、startWith、switch、zip
• 错误处理：catch、retry
• 辅助操作：delay、do、materialize/dematerialize、observeOn、serialize、subscribe、subscribeOn、timeInterval、timeout、timestamp、using、to
• 条件和布尔操作：all/contains/amb、defaultEmpty、SequenceEqual、skipUntil/skipWhile、takeUntil/TakeWhile
• 算数和聚合操作：average/concat/reduce、max/min/count/sum
• 异步操作：start、toAsync、、、
• 连接操作：connect、publish、refCount、replay
• 转换操作：toFuture、、、
• 阻塞操作：forEach、、、
• 字符串操作：byLine、docede、、、

2 使用案例解析

2.1 操作符just、map组合

2.1.1 场景描述

        输入网络图片路径path（String），通过网络请求获取原图Bitmap，然后裁剪Bitmap，并且转成Drawable。最后给文字添加背景图片Drawable。

2.1.2 主要代码

1.  
   private void doRxJava1(String path) {
2.  
   Log.d(TAG, "doRxJava1: start");
3.  
   Observable.just(path)
4.  
   // 需求1：获取图片
5.  
   .map(new Function<String, Bitmap>() {
6.  
   @Override
7.  
   public Bitmap apply(@NonNull String path) throws Exception {
8.  
   Log.d(TAG, "io线程 apply需求1：正在获取网络图片，需要3秒...");
9.  
   // String --> Bitmap
10.  
    Bitmap bitmap = obtainPictureByHttp(path);
11.  
    return bitmap;
12.  
    }
13.  
    })
14.  
    // 需求2：裁剪图片
15.  
    .map(new Function<Bitmap, Drawable>() {
16.  
    @Override
17.  
    public Drawable apply(@NonNull Bitmap bitmap) throws Exception {
18.  
    Log.d(TAG, "io线程 apply需求2：裁剪图片");
19.  
    // Bitmap --> Drawable
20.  
    Drawable drawable = cropPicture(bitmap);
21.  
    return drawable;
22.  
    }
23.  
    })
24.  
    // .map(todo 实现更多需求)
25.  
    .subscribeOn(Schedulers.io())
26.  
    .observeOn(AndroidSchedulers.mainThread())
27.  
    .subscribe(new Observer<Drawable>() {
28.  
    @Override
29.  
    public void onSubscribe(@NonNull Disposable d) {
30.  
    Log.d(TAG, "调用subscribe的线程 onSubscribe: ");
31.  
    }
32.  
     
33.  
    @Override
34.  
    public void onNext(@NonNull Drawable drawable) {
35.  
    Log.d(TAG, "main线程 onNext: 设置背景图片" drawable);
36.  
    mTextView.setBackground(drawable);
37.  
    }
38.  
     
39.  
    @Override
40.  
    public void onError(@NonNull Throwable e) {
41.  
    Log.d(TAG, "main线程 onError: ");
42.  
    }
43.  
     
44.  
    @Override
45.  
    public void onComplete() {
46.  
    Log.d(TAG, "main线程 onComplete: 过程结束");
47.  
    }
48.  
    });
49.  
    }
50.  
     
51.  
    //********************************************************************************
52.  
    //模拟耗时操作，获取网络图片：String --> Bitmap
53.  
    private Bitmap obtainPictureByHttp(String path) {
54.  
    try {
55.  
    Thread.sleep(3000);
56.  
    } catch (InterruptedException e) {
57.  
    e.printStackTrace();
58.  
    }
59.  
    Drawable drawable = getDrawable(R.drawable.picture1);
60.  
    Bitmap bitmap = ((BitmapDrawable) drawable).getBitmap();
61.  
    return bitmap;
62.  
    }
63.  
     
64.  
    //裁剪图片: Bitmap --> Drawable
65.  
    private Drawable cropPicture(Bitmap bitmap) {
66.  
    // ...
67.  
    Drawable drawable = new BitmapDrawable(bitmap);
68.  
    return drawable;
69.  
    }

2.1.3 测试结果

 2.1.4 解析

just(T...): 将传入的参数依次发送出来

map( ) — 对序列的每一项都应用一个函数来变换Observable发射的数据序列

2.2 操作符create

2.2.1 场景描述

        初次获取网络文本，并显示，3秒后，再次获取网络文本，并显示。

2.2.2 主要代码

1.  
   private void doRxJava2(String path) {
2.  
   Log.d(TAG, "doRxJava2: start");
3.  
   Observable.create(new ObservableOnSubscribe<String>() { // 所有泛型一致
4.  
   @Override
5.  
   public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
6.  
   Log.d(TAG, "io线程 apply操作1：正在获取网络文本text111");
7.  
   emitter.onNext("text111");
8.  
   Thread.sleep(3000);
9.  
    
10.  
    Log.d(TAG, "io线程 apply操作2：正在获取网络文本text222");
11.  
    emitter.onNext("text222");
12.  
    emitter.onComplete();
13.  
    }
14.  
    })
15.  
    .subscribeOn(Schedulers.io())
16.  
    .observeOn(AndroidSchedulers.mainThread())
17.  
    .subscribe(new Observer<String>() {
18.  
    @Override
19.  
    public void onSubscribe(@NonNull Disposable d) {
20.  
    Log.d(TAG, "调用subscribe的线程 onSubscribe: ");
21.  
    }
22.  
     
23.  
    @Override
24.  
    public void onNext(@NonNull String text) {
25.  
    Log.d(TAG, "main线程 onNext: 设置文本" text);
26.  
    mTextView.setText(text);
27.  
    }
28.  
     
29.  
    @Override
30.  
    public void onError(@NonNull Throwable e) {
31.  
    Log.d(TAG, "main线程 onError: ");
32.  
    }
33.  
     
34.  
    @Override
35.  
    public void onComplete() {
36.  
    Log.d(TAG, "main线程 onComplete: 过程结束");
37.  
    }
38.  
    });
39.  
    }

2.2.3 测试结果

2.3 操作符flatMap、filter、map组合

2.3.1 场景描述

        界面上有一个自定义的视图 imageCollectorView ，它的作用是显示多张图 片，并能使用 addImage(Bitmap) 方法来任意增加显示的图片。现在需要程序将一个给出的目录数 组 File[] folders 中每个目录下的 png 图片都加载出来并显示在 imageCollectorView 中。需要注意 的是，由于读取图片的这一过程较为耗时，需要放在后台执行，而图片的显示则必须在 UI 线程执行

2.3.2 主要代码（直接用java多线程）

1.  
   new Thread() {
2.  
   @Override
3.  
   public void run() {
4.  
   super.run();
5.  
   for (File folder : folders) {
6.  
   File[] files = folder.listFiles();
7.  
   for (File file : files) {
8.  
   if (file.getName().endsWith(".png")) {
9.  
   final Bitmap bitmap = getBitmapFromFile(file);
10.  
    getActivity().runOnUiThread(new Runnable() {
11.  
    @Override
12.  
    public void run() {
13.  
    imageCollectorView.addImage(bitmap);
14.  
    }
15.  
    });
16.  
    }
17.  
    }
18.  
    }
19.  
    }
20.  
    }.start();

2.3.3 主要代码（采用RxJava） 

1.  
   private void showPictures(File[] folders) {
2.  
   Observable.fromArray(folders)
3.  
   .flatMap(new Function<File, Observable<File>>() {
4.  
   @Override
5.  
   public Observable<File> apply(File file) {
6.  
   return Observable.fromArray(file.listFiles());
7.  
   }
8.  
   })
9.  
   .filter(new Predicate<File>() {
10.  
    @Override
11.  
    public boolean test(@NonNull File file) throws Exception {
12.  
    return file.getName().endsWith(".png");
13.  
    }
14.  
    })
15.  
    .map(new Function<File, Bitmap>() {
16.  
    @Override
17.  
    public Bitmap apply(File file) {
18.  
    return getBitmapFromFile(file);
19.  
    }
20.  
    })
21.  
    .subscribeOn(Schedulers.io())
22.  
    .observeOn(AndroidSchedulers.mainThread())
23.  
    .subscribe(new Observer<Bitmap>() {
24.  
    @Override
25.  
    public void onSubscribe(@NonNull Disposable d) {}
26.  
    @Override
27.  
    public void onNext(@NonNull Bitmap bitmap) {
28.  
    imageCollectorView.addImage(bitmap);
29.  
    }
30.  
    @Override
31.  
    public void onError(@NonNull Throwable e) {}
32.  
    @Override
33.  
    public void onComplete() { }
34.  
    });
35.  
    }

2.4 取消订阅

2.5 线程切换

observeOn()与subscribeOn()用于切换线程

2.6 Rxjava操作符（defer，compose，retryWhen）

Rxjava操作符（defer，compose，retryWhen）_浅浅清风的博客-CSDN博客上一篇博客讲了Retrofit的简单使用，应该看过的都基本了解我们公司这个服务器请求网络数据的流程，我来简单梳理一下： 第一次登录，创建cookiejar，请求服务器数据，保存accesstoken到本地 请求其他网络数据，使用已存在的cookiejar，传入本地保存的accesstoken基本上是这样，但是在实际操作中，用户可能登录之后，过了很久才去请求其他数据，这时候cookie已经失效就https://blog.csdn.net/Fancy_xty/article/details/52275735

2.7 emitter.onNext

2.7.1 Demo代码

1.  
   private void doRxJava4(String path) {
2.  
   Log.d(TAG, "doRxJava2: start");
3.  
   Observable.create(new ObservableOnSubscribe<String>() {
4.  
   @Override
5.  
   public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
6.  
   Log.d(TAG, "io线程 第一次发射 text111");
7.  
   emitter.onNext("text111");
8.  
    
9.  
   Log.d(TAG, "io线程 第二次发射 text222");
10.  
    emitter.onNext("text222");
11.  
     
12.  
    Log.d(TAG, "io线程 发射Complete信号");
13.  
    emitter.onComplete();
14.  
    }
15.  
    })
16.  
    .map(new Function<String, String>() {
17.  
    @Override
18.  
    public String apply(@NonNull String text) throws Exception {
19.  
    Log.d(TAG, "io线程 map " text);
20.  
    return text;
21.  
    }
22.  
    })
23.  
    .subscribeOn(Schedulers.io())
24.  
    .observeOn(AndroidSchedulers.mainThread())
25.  
    .subscribe(new Observer<String>() {
26.  
    @Override
27.  
    public void onSubscribe(@NonNull Disposable d) {
28.  
    Log.d(TAG, "main线程 onSubscribe: ");
29.  
    }
30.  
     
31.  
    @Override
32.  
    public void onNext(@NonNull String text) {
33.  
    Log.d(TAG, "main线程 onNext " text);
34.  
    mTextView.setText(text);
35.  
    }
36.  
     
37.  
    @Override
38.  
    public void onError(@NonNull Throwable e) {
39.  
    Log.d(TAG, "main线程 onError: ");
40.  
    }
41.  
     
42.  
    @Override
43.  
    public void onComplete() {
44.  
    Log.d(TAG, "main线程 onComplete: 过程结束");
45.  
    }
46.  
    });
47.  
    }

2.7.2 测试结果

2.8 observer.onNext

2.8.1 Demo代码

1.  
   private void doRxJava4(String path) {
2.  
   Log.d(TAG, "doRxJava2: start");
3.  
   Observable.create(new ObservableOnSubscribe<String>() {
4.  
   @Override
5.  
   public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
6.  
   Log.d(TAG, "io线程 第一次发射 text111");
7.  
   emitter.onNext("text111");
8.  
    
9.  
   Log.d(TAG, "io线程 第二次发射 text222");
10.  
    emitter.onNext("text222");
11.  
     
12.  
    Log.d(TAG, "io线程 发射Complete信号");
13.  
    emitter.onComplete();
14.  
    }
15.  
    })
16.  
    .flatMap(new Function<String, ObservableSource<String>>() {
17.  
    @Override
18.  
    public ObservableSource<String> apply(@NonNull String text) throws Exception {
19.  
    Log.d(TAG, "io线程 flatMap apply " text);
20.  
    return new Observable<String>() {
21.  
    @Override
22.  
    protected void subscribeActual(Observer<? super String> observer) {
23.  
    Log.d(TAG, "io线程 observer.onNext " text);
24.  
    observer.onNext(text);
25.  
    observer.onNext(text);
26.  
    }
27.  
    };
28.  
    }
29.  
    })
30.  
    .subscribeOn(Schedulers.io())
31.  
    .observeOn(AndroidSchedulers.mainThread())
32.  
    .subscribe(new Observer<String>() {
33.  
    @Override
34.  
    public void onSubscribe(@NonNull Disposable d) {
35.  
    Log.d(TAG, "main线程 onSubscribe: ");
36.  
    }
37.  
     
38.  
    @Override
39.  
    public void onNext(@NonNull String text) {
40.  
    Log.d(TAG, "main线程 onNext " text);
41.  
    mTextView.setText(text);
42.  
    }
43.  
     
44.  
    @Override
45.  
    public void onError(@NonNull Throwable e) {
46.  
    Log.d(TAG, "main线程 onError: ");
47.  
    }
48.  
     
49.  
    @Override
50.  
    public void onComplete() {
51.  
    Log.d(TAG, "main线程 onComplete: 过程结束");
52.  
    }
53.  
    });
54.  
    }

2.8.2 测试结果

这篇好文章是转载于：学新通技术网