Home 初学RxJava
Post
Cancel

初学RxJava

什么是RxJava?

这个名词包含两部分:

  • Rx(是ReactiveX、Reactive Extensions、Reactive Programming的简称):An API for asynchronous programming with observable streams.
  • Java: Rx在JVM上实现。

在RxJava的GitHub里官方给出了这样的示意:

A library for composing asynchronous and event-based programs by using observable sequences.

Rx组合了观察者模式迭代器模式函数式编程范式思想,所以有时候也将这种编程方法叫做Functional Reactive Programming

Hello world

1
2
3
4
5
6
7
8
Observable.just("Hello", "world!")
	    .subscribe(new Consumer<String>() {
	        @Override
	        public void accept(String item) throws Exception {
	            System.out.print(item + " ");
	        }
	    });
// Hello world!

Observable.just()将传入的2个字符串项转化成一个Reactive Stream,调用subscribe()订阅方法观察该Stream里的数据项。图示如下:

image.png

Observable和Observer

Observable是一种基于事件推送并且可组合的迭代器,它可以推送3种事件:

  • onNext():接收推送过来的数据项。
  • onError():接收失败事件。
  • onComplete():响应流推送已完成。

这3种事件与Observer类中的抽象方法一一对应。

1
2
3
4
5
public interface Observer<T> {
    void onNext(@NonNull T t);
    void onError(@NonNull Throwable e);
    void onComplete();
}

使用Observer订阅Observable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
Observable.just("Hello", "world!")
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String item) {
                out.print(item + " ");
            }

            @Override
            public void onError(Throwable e) {
                out.println("error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                out.println("done");
            }
        });

Observer.onSubscribe()方法的入参Disposable可用来取消订阅。下面使用RxJava提供的DisposableObserver类演示取消订阅功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
Disposable disposable = Observable.just("Hello", "world!")
        .subscribeWith(new DisposableObserver<String>() {
            @Override
            public void onNext(String item) {
                out.print(item + " ");
            }

            @Override
            public void onError(Throwable e) {
                out.println("error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                out.println("done");
            }
        });

// 在适当的时机取消订阅
disposable.dispose();

Operator

Observable.just()也是一种Operator,它将传入的数据项转化成了响应流。对响应流应用一些Operator可以进行函数式编程、数据转换、过滤等许多功能。

map

image.png

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Observable.just("Hello", "world!")
        .map(new Function<String, Integer>() {
            @Override
            public Integer apply(String item) throws Exception {
                return item.length();
            }
        })
        .subscribe(new DisposableObserver<Integer>() {
            @Override
            public void onNext(Integer item) {
                out.print(item + " ");
            }

            @Override
            public void onError(Throwable e) {
                out.println("error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                out.println("done");
            }
        });

create

使用create()方法可以自己动手创建响应流。

image.png

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                try {
                    e.onNext("Hello");
                    e.onNext(" world!");
                    e.onComplete();
                } catch (Exception ex) {
                    e.onError(ex);
                }
            }
        })
        .subscribe(new DisposableObserver<String>() {
            @Override
            public void onNext(String item) {
                out.print(item + " ");
            }

            @Override
            public void onError(Throwable e) {
                out.println("error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                out.println("done");
            }
        }); 

使用场景

异步操作

作为一个Android开发者,我们都知道:不应该阻塞UI线程,应该将耗时的计算放在异步线程中。我们有多种实现方式,这里不再多做说明,直接使用RxJava实现异步操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            // 该方法将在IO线程里执行
            try {
                e.onNext("Hello");
                e.onNext(" world!");
                e.onComplete();
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new DisposableObserver<String>() {
        @Override
        public void onNext(String item) {
            // 在UI线程里执行
            out.print(item + " ");
        }

        @Override
        public void onError(Throwable e) {
            // 在UI线程里执行
            out.println("error: " + e.getMessage());
        }

        @Override
        public void onComplete() {
            // 在UI线程里执行
            out.println("done");
        }
    });

多异步操作

现在要实现一个修改头像的需求,分成以下3个步骤来完成:

  • 获取图片上传token;
  • 使用上传token上传图片;
  • 调用修改个人信息接口修改头像信息。

很显然这3个操作都需要在异步线程里进行,而且是按顺序一一进行的。通常使用回掉来实现非阻塞异步操作,但随着连续多个异步操作容易引入Callback Hell问题。

使用RxJava的flatMap()方法可以优雅地解决这些问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
getUploadToken()
    .flatMap(token -> uploadImage(token))
    .flatMap(imageUrl -> updateUser(imageUrl))
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new DisposableObserver<Boolean>() {
        @Override
        public void onNext(Boolean item) {
            out.print(item + " ");
        }

        @Override
        public void onError(Throwable e) {
            out.println("error: " + e.getMessage());
        }

        @Override
        public void onComplete() {
            out.println("done");
        }
    });

public Observable<String> getUploadToken() {
    return Observable.create(e -> {
        try {
            String token = "get token from remote server";
            e.onNext(token);
            e.onComplete();
        } catch (Exception ex) {
            e.onError(ex);
        }
    });
}

public Observable<String> uploadImage(final String token) {
    return Observable.create(e -> {
        try {
            String imageUrl = "upload image to remote server with token: " + token;
            e.onNext(imageUrl);
            e.onComplete();
        } catch (Exception ex) {
            e.onError(ex);
        }
    });
}

public Observable<Boolean> updateUser(final String imageUrl) {
    return Observable.create(e -> {
        try {
            // update user with image url
            e.onNext(true);
            e.onComplete();
        } catch (Exception ex) {
            e.onError(ex);
        }
    });
}

image.png

防止连续点击

在Android开发中会遇到这样的问题:短时间内快速点击某个按钮,按钮的单击监听器会被多次调用,这样会引入一写异常情况。

同样可以使用RxJava的throttleFirst()方法来解决这种问题。

image.png

1
2
3
4
5
RxView.clicks(findViewById(R.id.btn_throttle))
    .throttleFirst(1, TimeUnit.SECONDS)
    .subscribe(aVoid -> {
        System.out.println("click");
    });

参考资料

This post is licensed under CC BY 4.0 by the author.