This post is from a low-reputation account and contains an unverified outbound link. Be cautious before clicking external links.

RxJava2.0操作符之--创建操作符

hiquanta(3)
Published in
#steemit
Words
923
Reading
5 min
9y
Listen to this post
~6 min

之前写过几遍RxJava相关的博客，但都是基于1.x版本的，2.x出来也很久了，所以用2.x把之前写的更新下
可能RxJava里面最重要也最难理解的就是操作符了，理解是第一步，理解后怎么用又是一个问题，接下来的几篇先把理解操作符这个问题搞定，然后我会将一讲相关的应用场景。相关操作符的解释官网都有，但是相关的实例代码不是很全，很多操作符只是一行解释而已，我决定将文档提到的常用的操作符都一一用代码展示出来，更直接明了,也算是对文档的补充

Create

使用一个函数从头开始创建一个Observable

先创建被观察者：

Observable observable=   Observable.create(new ObservableOnSubscribe()
   {
       public void subscribe(@NonNull ObservableEmitter observableEmitter) throws Exception {

           observableEmitter.onNext(0);
  observableEmitter.onNext(1);
  observableEmitter.onNext(2);
  observableEmitter.onNext(3);
  observableEmitter.onNext(4);
  observableEmitter.onNext(5);
  observableEmitter.onComplete();
  }
   });

然后是观察者（同一个事件可以被多个观察者接收）,这里对创建的代码进行了封装，创建出两个观察者

public static Observer getObserver(final int observableId){
    return new Observer() {
        public void onSubscribe(@NonNull Disposable disposable) {
            System.out.println("onSubscribe:"+observableId);
  }

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

        public void onError(@NonNull Throwable throwable) {
            System.out.println("onError:"+throwable);
  }

        public void onComplete() {
            System.out.println("onComplete");
  }
    };
}
    Observer observer=getObserver(1);
    Observer observer2=getObserver(2);

最后订阅时间流，也就是关联绑定观察者与被观察者

    observable.subscribe(observer);
    observable.subscribe(observer2);

运行结果：

onSubscribe:1
Thread:Thread[main,5,main]
onNext:0
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]
onSubscribe:2
Thread:Thread[main,5,main]
onNext:0
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]

可以看到create方法默认不在任何特定的调度器上执行。

defer

直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable

创建关联过程和上面的类似，都是基于观察者模式的创建方式，我就不详细叙述，直接贴代码

    i = 10;
  Observable justObservable = Observable.just(i);
  i = 12;
  Observable deferObservable=Observable.defer(new Callableextends Integer>>() {
      public ObservableSourceextends Integer> call() throws Exception {
          return Observable.just(i);
    }
  });
  i=15;
  justObservable.subscribe(RxUtils.getObserver(1));
  deferObservable.subscribe(RxUtils.getObserver(2));

运行结果:

    onSubscribe:1
  Thread:Thread[main,5,main]
  onNext:10
  Thread:Thread[main,5,main]
  onComplete
  Thread:Thread[main,5,main]
  onSubscribe:2
  Thread:Thread[main,5,main]
  onNext:15
  Thread:Thread[main,5,main]
  onComplete
  Thread:Thread[main,5,main]

通过运行结果可以很明显的看出，defer操作符的特征

empty

创建一个不发射任何数据但是正常终止的Observable

  Observable emptyObservable= Observable.empty();
  emptyObservable.subscribe(RxUtils.getObserver(0));

运行结果

    Observable emptyObservable= Observable.empty();
    emptyObservable.subscribe(RxUtils.getObserver(0));

never

创建一个不发射数据也不终止的Observable

  Observable nerverObservable= Observable.never();
  nerverObservable.subscribe(RxUtils.getObserver(0));

运行结果

  onSubscribe:0
  Thread:Thread[main,5,main]

error

创建一个不发射数据以一个错误终止的Observable

  Observable errorObservable=Observable.error(new Throwable());
  errorObservable.subscribe(RxUtils.getObserver(0));

运行结果:

  Observable errorObservable=Observable.error(new Throwable());
  errorObservable.subscribe(RxUtils.getObserver(0));

fromXXX

将其它种类的对象和数据类型转换为Observable

    Integer[] items = { 0, 1, 2, 3, 4, 5 };
    Observable fromObservable = Observable.fromArray(items);
    fromObservable.subscribe(RxUtils.getObserver(0));

运行结果：

    onSubscribe:0
    Thread:Thread[main,5,main]
    onNext:0
    Thread:Thread[main,5,main]
    onNext:1
    Thread:Thread[main,5,main]
    onNext:2
    Thread:Thread[main,5,main]
    onNext:3
    Thread:Thread[main,5,main]
    onNext:4
    Thread:Thread[main,5,main]
    onNext:5
    Thread:Thread[main,5,main]
    onComplete
    Thread:Thread[main,5,main]

interval

创建一个按固定时间间隔发射整数序列的Observable

    Observable observable= Observable.interval(1, TimeUnit.SECONDS);
    observable.subscribe(RxUtils.getObserver(0));
    try {
        Thread.sleep(Integer.MAX_VALUE);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

运行结果

onSubscribe:0
Thread:Thread[main,5,main]
onNext:0
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:6
Thread:Thread[RxComputationThreadPool-1,5,main]
......

just

创建一个发射指定值的Observable

  Observable.just(1, 2, 3).subscribe(RxUtils.getObserver(0));

运行结果

onSubscribe:0
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]

range

创建一个发射特定整数序列的Observable

  Observable.range(3, 10).subscribe(RxUtils.getObserver(0));

运行结果

  onSubscribe:0
  Thread:Thread[main,5,main]
  onNext:3
  Thread:Thread[main,5,main]
  onNext:4
  Thread:Thread[main,5,main]
  onNext:5
  Thread:Thread[main,5,main]
  onNext:6
  Thread:Thread[main,5,main]
  onNext:7
  Thread:Thread[main,5,main]
  onNext:8
  Thread:Thread[main,5,main]
  onNext:9
  Thread:Thread[main,5,main]
  onNext:10
  Thread:Thread[main,5,main]
  onNext:11
  Thread:Thread[main,5,main]
  onNext:12
  Thread:Thread[main,5,main]
  onComplete
  Thread:Thread[main,5,main]

repeat

repeat操作符就是对某一个Observable重复产生多次结果,当repeat()

接收到onComplete()会触发重订阅，默认情况下运行在一个新的线程上

  Observable.range(1, 5).repeat(5).subscribe(RxUtils.getObserver(0));

onSubscribe:0
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onComplete
Thread:Thread[main,5,main]

repeatWhen

它不是缓存和重放原始Observable的数据序列，而是有条件的重新订阅和发射原来的Observable。

将原始Observable的终止通知（完成或错误）当做一个void数据传递给一个通知处理器，

它以此来决定是否要重新订阅和发射原来的Observable。这个通知处理器就像一个Observable操作

符，接受一个发射void通知的Observable为输入，返回一个发射void数据（意思是，重新订阅和发射原始Observable）或者

直接终止（意思是，使用repeatWhen终止发射数据）的Observable。

Observable.range(1, 5).repeatWhen(new Function, ObservableSource>() {
    public ObservableSource apply(@NonNull Observable objectObservable) throws Exception {
        //延时六秒发送数据
  return objectObservable.delay(6, TimeUnit.SECONDS);
  }
}).subscribe(RxUtils.getObserver(0));
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e) {
    e.printStackTrace();
}

运行结果

onSubscribe:0
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[main,5,main]
onNext:2
Thread:Thread[main,5,main]
onNext:3
Thread:Thread[main,5,main]
onNext:4
Thread:Thread[main,5,main]
onNext:5
Thread:Thread[main,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:2
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:3
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:4
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:5
Thread:Thread[RxComputationThreadPool-1,5,main]
onNext:1

timer

创建一个Observable，它在一个给定的延迟后发射一个特殊的值。

    Observable.timer(4, TimeUnit.SECONDS).subscribe(RxUtils.getObserver(Long.valueOf(0)));
    try {
        Thread.sleep(Integer.MAX_VALUE);
    } catch (InterruptedException e1) {
        e1.printStackTrace();
    }

运行结果

onSubscribe:0
Thread:Thread[main,5,main]
onNext:0
Thread:Thread[RxComputationThreadPool-1,5,main]
onComplete
Thread:Thread[RxComputationThreadPool-1,5,main]

RxJava的常用创建操作符就介绍这么多，需要注意的是观察和注意下每个操作符线程的调度（log输出里看），体会一下，晚上我会把代码发到github
当然这里面有些需要注意的细节我会慢慢完善。

steemit
cn
news
java
cn-programming
via steemit
This post was published via steemit. Ecency is not the originator or editor of this content and displays it for discovery purposes only.
1
RxJava2.0操作符之--创建操作符 | Ecency