Rxjava 中 merge 和 contact 区别，然后co

contact 连接

concat 有道.png concat.png

看到没有，是一个接一个的，用图来说就是

contact

例如，依次检查memory、disk和network中是否存在数据，任何一步一旦发现数据后面的操作都不执行。

Observable<String> memory = Observable.create(new Observable.OnSubscribe<String>() {  
    @Override  
    public void call(Subscriber<? super String> subscriber) {  
        if (memoryCache != null) {  
            subscriber.onNext(memoryCache);  
        } else {  
            subscriber.onCompleted();  
        }  
    }  
});  
Observable<String> disk = Observable.create(new Observable.OnSubscribe<String>() {  
    @Override  
    public void call(Subscriber<? super String> subscriber) {  
        String cachePref = rxPreferences.getString("cache").get();  
        if (!TextUtils.isEmpty(cachePref)) {  
            subscriber.onNext(cachePref);  
        } else {  
            subscriber.onCompleted();  
        }  
    }  
});  
  
Observable<String> network = Observable.just("network");  
  
//依次检查memory、disk、network  
Observable.concat(memory, disk, network)  
.first()  //这里可以做缓存过期监测
.subscribeOn(Schedulers.newThread())  
.subscribe(s -> {  
    memoryCache = "memory";  
    System.out.println("--------------subscribe: " + s);  
});  

对于contact，有一点务必要注意，那就是上一个数据源不可用，请务必subscriber.onCompleted(); 这样才能继续走下一个数据源。

上面提到了缓存过期监测，那么举个🌰吧。

Observable<Data> source = Observable.concat(        
  sources.memory(),        
  sources.disk(),        
  sources.network()  )    
.first(data -> data != null && data.isUpToDate());//lamada表达式而已，别怕

让我们看一下first的定义吧，当然如果first什么参数不传，那就直接第一个了。

first.png

merge 合并

merge

用图来说话就是：

merge

例如一组数据来自网络，一组数据来自文件，需要合并两组数据一起展示。

Observable.merge(getDataFromFile(), getDataFromNet())  
              .observeOn(AndroidSchedulers.mainThread())  
              .subscribe(new Subscriber<String>() {  
                  @Override  
                  public void onCompleted() {  
                      log.d("done loading all data");  
                  }  
  
                  @Override  
                  public void onError(Throwable e) {  
                      log.d("error");  
                  }  
  
                  @Override  
                  public void onNext(String data) {  
                      log.d("all merged data will pass here one by one!")  
              });  

compose 组合

可以将一组变换用到流上，例如抛物线写的那个例子

compose
给 Android 开发者的 RxJava 详解
以及这篇文章，我想也可以帮助你。
http://blog.danlew.net/2015/03/02/dont-break-the-chain/