首先我们创建两个Observable1 Observable2来模拟网络请求,代码如下:
//每个子Observable都要加subscribeOn(Schedulers.io())指定为异步线程 不同线程来处理请求达到并发
val observable1: Observable<String> = Observable.create(ObservableOnSubscribe<String> {
Log.e("asd" , "observable1---"+Thread.currentThread().name)
Thread.sleep(5000)
it.onNext("1111")
it.onComplete()
}).subscribeOn(Schedulers.io())
val observable2: Observable<String> = Observable.create(ObservableOnSubscribe<String>{
Log.e("asd" , "observable2---"+Thread.currentThread().name)
Thread.sleep(10000)
it.onNext("2222")
it.onComplete()
}).subscribeOn(Schedulers.io())
observable1中第一行打印了线程的名字,当执行到这里的时候代表该请求已经在执行了,然后让线程休眠5秒钟模拟耗时,休眠过后发送一个111字符串出去调用onComplete完成本此请求
observable2中类似和observable1类似,只是休眠时间变成10秒,10秒过后请求完成发送数据,因为两个请求由于返回数据和网络状态,等等原因,他们所消耗的时间肯定不一样的。
这里需要注意 每个子的Observable必须要单独指定线程subscribeOn(Schedulers.io()),如果不加这个,那么两个请求还是在一个线程中执行,必定会有先后,达不到并发的效果,这里后面会展示不并发的效果
接下来吧请求合并一起使用zip关键字代码如下:
val start = System.currentTimeMillis()
Observable.zip(observable1, observable2,BiFunction<String, String, String> { t1, t2 ->
Log.e("asd" , "zip:"+Thread.currentThread().name)
//这里是两个请求都返回了之后 执行到这里 把数据封装到一个类中返回,这里只是简单的吧他们加起来返回了
t1 + t2
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe {
val end = System.currentTimeMillis() - start
Log.e("asd" , "请求的数据:${it}---耗时:${end}---observable:"+Thread.currentThread().name)
}
看下打印的日志,分析一下
2019-10-25 17:18:39.143 19436-19772/com.joyrun.study E/asd: observable2---RxCachedThreadScheduler-3
2019-10-25 17:18:39.143 19436-19771/com.joyrun.study E/asd: observable1---RxCachedThreadScheduler-2
2019-10-25 17:18:49.144 19436-19772/com.joyrun.study E/asd: zip:RxCachedThreadScheduler-3
2019-10-25 17:18:49.145 19436-19436/com.joyrun.study E/asd: 请求的数据:11112222---耗时:10004---observable:main
17:18:39 同一时间打印了observable1 observable2,说明这两个请求已经并发执行了
17:18:49 10秒钟过后两个请求都完成了,同一时间把数据返回主线程,打印出请求的数据:11112222---耗时:10004---observable:main
总共耗时10004 10秒 也就是说两个请求并发了,最后拿到数据的消耗时间是俩个请求中最长的耗时时间,如果你想先执行完 就先处理的话 使用merge操作符就可以了,最后面展示用法
下面看下 不并发的情况,也就是每个子Observable不单独指定线程,那么他们还是在同一线程中执行,打印日志中可以看到,所以必定是有一个先一个后执行的,达不到并发效果
val observable1: Observable<String> = Observable.create {
Log.e("asd" , "observable1:"+Thread.currentThread().name)
Thread.sleep(5000)
it.onNext("1111")
it.onComplete()
}
val observable2: Observable<String> = Observable.create {
Log.e("asd" , "observable2:"+Thread.currentThread().name)
Thread.sleep(10000)
it.onNext("2222")
it.onComplete()
}
解释:使用Observable.zip将两个Observable合并成一个指定子线程,然后在BiFunction的接口函数中就可以拿到这两个请求返回的数据,主意执行到这里一定是两个请求都成功返回数据了,看下打印的日志
2019-10-25 17:03:46.903 11409-11465/com.joyrun.study E/asd: observable1:RxCachedThreadScheduler-1
2019-10-25 17:03:51.905 11409-11465/com.joyrun.study E/asd: observable2:RxCachedThreadScheduler-1
2019-10-25 17:04:01.906 11409-11465/com.joyrun.study E/asd: zip:RxCachedThreadScheduler-1
2019-10-25 17:04:01.907 11409-11409/com.joyrun.study E/asd: 请求的数据:11112222---耗时:15006---observable:main
17:03:46 开始了第一个请求
17:03:51 五秒过后执行了第二个请求
17:04:01 也就是10秒过后第二个请求执行完成 就会调用到BiFunction的接口函数中做两个请求数据的处理并返回
17:04:01 同一时间将数据返回到了主线程
打两个请求合并之后的请求总时间15006 15秒钟 一个5秒一个10秒,虽然请求合并了 但是没有并发 请求所消耗的时间 还是两个请求消耗时间的总和
如果说想并发请求,那个返回数据 就处理那个的话就是用merge操作符即可低吗如下
Observable.merge(observable1, observable2).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()).subscribe {
Log.e("asd" , "请求的数据:${it}---observable:"+Thread.currentThread().name)
}
打印日志
2019-10-25 17:28:48.586 25014-25108/com.joyrun.study E/asd: observable2:RxCachedThreadScheduler-3
2019-10-25 17:28:48.587 25014-25107/com.joyrun.study E/asd: observable1:RxCachedThreadScheduler-2
2019-10-25 17:28:53.589 25014-25014/com.joyrun.study E/asd: 请求的数据:1111---observable:main
2019-10-25 17:28:58.587 25014-25014/com.joyrun.study E/asd: 请求的数据:2222---observable:main
通过打印日志看出两个请求并发执行,并且在不同线程中
17:28:53 5秒后执行完第一个请求拿到返回数据进行处理
17:28:58 10秒后执行完第二个请求,拿到返回数据进行处理,注意10秒后 是针对两个请求并发处理之后的10秒 并不是第一个请求完了 后10秒
测试中完整代码如下:
//----------------------------------------------------------
// val observable1: Observable<String> = Observable.create {
//// Log.e("asd" , "observable1:"+Thread.currentThread().name)
//// Thread.sleep(5000)
//// it.onNext("1111")
//// it.onComplete()
//// }
//// val observable2: Observable<String> = Observable.create {
//// Log.e("asd" , "observable2:"+Thread.currentThread().name)
//// Thread.sleep(10000)
//// it.onNext("2222")
//// it.onComplete()
//// }
//----------------------------------------------------------
//**********************************************************
//每个子Observable都要加subscribeOn(Schedulers.io())指定为异步线程 不同线程来处理请求达到并发
val observable1: Observable<String> = Observable.create(ObservableOnSubscribe<String> {
Log.e("asd" , "observable1:"+Thread.currentThread().name)
Thread.sleep(5000)
it.onNext("1111")
it.onComplete()
}).subscribeOn(Schedulers.io())
val observable2: Observable<String> = Observable.create(ObservableOnSubscribe<String>{
Log.e("asd" , "observable2:"+Thread.currentThread().name)
Thread.sleep(10000)
it.onNext("2222")
it.onComplete()
}).subscribeOn(Schedulers.io())
//**********************************************************
//zip
// val start = System.currentTimeMillis()
// Observable.zip(observable1, observable2,BiFunction<String, String, String> { t1, t2 ->
// Log.e("asd" , "zip:"+Thread.currentThread().name)
// //这里是两个请求都返回了之后 执行到这里 把数据封装到一个类中返回,这里只是简单的吧他们加起来返回了
// t1 + t2
// }).subscribeOn(Schedulers.io())
// .observeOn(AndroidSchedulers.mainThread())
// .subscribe {
// val end = System.currentTimeMillis() - start
// Log.e("asd" , "请求的数据:${it}---耗时:${end}---observable:"+Thread.currentThread().name)
// }
//merge
Observable.merge(observable1, observable2).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()).subscribe {
Log.e("asd" , "请求的数据:${it}---observable:"+Thread.currentThread().name)
}