Rxjava笔记

Author Avatar
w4ctech 3月20日
  • 在其它设备中阅读本文章

最后更新于2019年03月21日; 如遇到问题,请留言及时通知站长

添加依赖

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.1'

compile 'io.reactivex.rxjava2:rxandroid:2.1.0'
compile 'io.reactivex.rxjava2:rxjava:2.2.3'

创建数据源(被观察者 / 可观察者)

just 方法
返回一个可观察对象,该对象发出给定 (常量引用) 项的信号,然后完成
Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("just:" + integer);
    }
});

just:1
just:2
just:4
just:5
just:6
just:7
just:8
just:9
just:10
fromArray 方法
fromArray 和 Just 几乎是一样的效果,只不过 Just 限制 10 个以内,而 fromArray 并没有限制,查看得知 单个参数 Just 是自行创建 ObservableJust,而多个参数 Just 最终还是回调了 fromArray,这里不再过多演示
create 方法
Observable.create(new ObservableOnSubscribe<Integer>() {

    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        //执行多次
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        //标记事件结束
        emitter.onComplete();
        //标记事件发送错误
        //emitter.onError(new NullPointerException("不能为空"));
    }
}).subscribe(new Consumer<Integer>() {

    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("create:" + integer);
    }
});

create:1
create:2
create:3
range 方法
使用范围数据,指定输出数据的范围(1-40 的数值)
Observable.range(3, 5)
.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("range:" + integer);
    }
});

range:3
range:4
range:5
range:6
range:7
interval 方法
指定某一时刻进行数据发送
Observable.interval(10, 1, TimeUnit.SECONDS) // 先等待 10 秒,之后再每一秒发送一次,10 秒这个参数也可以不填,默认用间隔时间参数替代(这里示例 1 秒)
.subscribe(new Consumer<Long>() {
    @Override
    public void accept(Long l) throws Exception {
        System.out.println("interval:" + l);
    }
});

2019-03-20 15:30:33.331 interval:0
2019-03-20 15:30:34.331 interval:1
2019-03-20 15:30:35.331 interval:2
2019-03-20 15:30:36.331 interval:3
2019-03-20 15:30:37.331 interval:4
2019-03-20 15:30:38.331 interval:5
..................................

创建事件的接收者(观察者 | 订阅者),onNext 方法中的数据类型必须被观察者指定的泛型

// 表示观察者不对被观察者发送的事件作出任何响应(但被观察者还是可以继续发送事件)
public final Disposable subscribe() {}

// 表示观察者只对被观察者发送的Next事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext) {}

// 表示观察者只对被观察者发送的Next事件 & Error事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 

// 表示观察者只对被观察者发送的Next事件、Error事件 & Complete事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}

// 表示观察者只对被观察者发送的Next事件、Error事件 、Complete事件 & onSubscribe事件作出响应
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}

// 表示观察者对被观察者所有的事件做出响应
public final void subscribe(Observer<? super T> observer) {}
Observer 类用法
onSubscribe:订阅的时候被调用,方法参数有 Disposable,可用于取消订阅

onNext(T item):Observable 调用这个方法发射数据,方法的参数就是 Observable 发射的数据,这个方法可能会被调用多次,取决于业务逻辑

onCompleted():正常终止,在没有遇到错误的情况下,Observable 在最后一次调用 onNext 之后调用此方法

onError(Throwable e):当 Observable 遇到错误或者无法返回期望的数据时会调用这个方法,这个调用会终止 Observable,后续不会再调用 onNext 和 onCompleted,onError 方法的参数是抛出异常

//观察者
Observable.just(1, 2, 3)
.subscribe(new Observer<Integer>() {

    @Override
    public void onSubscribe(Disposable d) {
        System.out.println("onSubscribe");
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext:" + integer);
    }

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

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

onSubscribe
onNext:1
onNext:2
onNext:3
onComplete
Consumer 类用法
Observable.just(1, 2, 3)
.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println("accept:" + integer);
    }
});

accept:1
accept:2
accept:3

订阅

// 订阅事件,被观察者必须指定了事件的接收者(观察者),整个事件流程才可以被启动
Disposable disposable = observable.subscribe(observer);

// 是否被订阅
disposable.isDisposed();

// 取消订阅
disposable.dispose();

本文链接:https://i.w4ctech.cn/Rxjava/note2.html
This blog is under a CC BY-NC-SA 3.0 Unported License