RxJava2.0操作符之--连接操作符

Connect

让一个可连接的Observable开始发射数据给订阅者

publish

将普通的Observable转换为可连接的Observable

  Observable observable=Observable.just(1,2,3);
ConnectableObservable connectableObservable = observable.publish();
connectableObservable.subscribe(RxUtils.getObserver());
connectableObservable.connect();
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}

onSubscribe
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]

refCount

让一个可连接的Observable行为像普通的Observable

Observable observable=Observable.just(1,2,3);
  ConnectableObservable connectableObservable = observable.publish();
  Observable observable1=  connectableObservable.refCount();
  observable1.subscribe(RxUtils.getObserver());
//   connectableObservable.connect();
  try {
       Thread.sleep(Integer.MAX_VALUE);
  } catch (InterruptedException e1) {
       e1.printStackTrace();
  }

    onSubscribe
    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]

Replay

保证所有的观察者收到相同的数据序列，即使它们在Observable开始发射数据之后才订阅

2.x 代码中没有达到预期，1.x却有效 不知道为什么,又在2.x成功的大神，给说下，谢谢

Observable observable=Observable.interval(1, TimeUnit.SECONDS).take(5);
ConnectableObservable connectableObservable = observable.publish();
connectableObservable
     .replay(4) //缓存4个数据
  .publish();
connectableObservable.connect();
connectableObservable
       .delaySubscription(4,TimeUnit.SECONDS)
        .subscribe(RxUtils
                .getObserver());
try {
    Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException e1) {
    e1.printStackTrace();
}