Rxjava3使用教程：操作符-创建

创建操作符

名称	描述
create()	创建最简单的事件流
from()	创建事件流，可发送不同类型的数据流
just()	创建事件流，可发送多个参数的数据流
range()	创建事件流，可发送范围内的数据流
repeat()	创建可重复次数的事件流
defer()	创建事件流，可缓存可激活事件流
interval()	创建延时重复的事件流
timer()	创建一次延时的事件流

tips: defer、interval、timer之间的区别

• defer: 用于事件流中，可以延时某次事件流的发送
• interval: 用于创建事件流，周期性重复发送
• timer: 用于创建事件流，延时发送一次

1. create

使用方式

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onError(new NullPointerException("null..."));
        emitter.onComplete();

    }
});
observable.subscribe(observer);

2. from

使用方式

Observable.fromArray(1,2,3,4,5).subscribe(observer);

3. just

使用方式

Observable.just(1,3,5).subscribe(observer);

4. range

使用方式

Observable.range(1, 10).subscribe(observer);

5. repeat

使用方式

Observable.fromArray(1,3,5,7,9).repeat(5).subscribe(observer);

6. defer

使用方式

Observable.defer(() -> Observable.create(emitter -> {
  emitter.onNext(10);
  emitter.onComplete();
})).subscribe(aLong -> {
  Observable.just(1,3,5).subscribe(observer);
});

7. interval

使用方式

Observable.interval(2, TimeUnit.SECONDS).subscribe(aLong -> {
  Observable.just(1,3,5).subscribe(observer);
});

8. timer

使用方式

Observable.timer(5, TimeUnit.SECONDS).subscribe(aLong -> {
  Observable.just(1,3,5).subscribe(observer);
});

9. observer

private Observer<Integer> observer = new Observer<Integer>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
        Log.d(TAG, "onSubscribe:");
    }

    @Override
    public void onNext(@NonNull Integer integer) {
        Log.d(TAG, "onNext:" + integer);
    }

    @Override
    public void onError(@NonNull Throwable e) {
        Log.d(TAG, "onError:" + Log.getStackTraceString(e));
    }

    @Override
    public void onComplete() {
        Log.d(TAG, "onComplete");
    }
};