节流

先来看一下例子,当使用concat进行合流操作时，我们打印时如下。

  val source1 = Observable.interval(100, TimeUnit.MILLISECONDS)
                .map { i -> (i + 1) * 100 } // map to elapsed time
                .map { i -> "SOURCE 1: $i" }
                .take(10)
        val source2 = Observable.interval(300, TimeUnit.MILLISECONDS)
                .map { i -> (i + 1) * 300 } // map to elapsed time
                .map { i -> "SOURCE 2: $i" }
                .take(3)
        val source3 = Observable.interval(2000, TimeUnit.MILLISECONDS)
                .map { i -> (i + 1) * 2000 } // map to elapsed time
                .map { i -> "SOURCE 3: $i" }
                .take(2)
        Observable.concat(source1, source2, source3)
                .subscribe{
                    println(it)
                }
        sleep(6000)
        /**
SOURCE 1: 100
SOURCE 1: 200
SOURCE 1: 300
SOURCE 1: 400
SOURCE 1: 500
SOURCE 1: 600
SOURCE 1: 700
SOURCE 1: 800
SOURCE 1: 900
SOURCE 1: 1000
SOURCE 2: 300
SOURCE 2: 600
SOURCE 2: 900
SOURCE 3: 2000
SOURCE 3: 4000
        */

如果我们想要只发送他们的最后一个怎么做呢

• throttleLatest

  Observable.concat(source1, source2, source3)
                .throttleLatest(1, TimeUnit.SECONDS)
                .subscribe {
                    println(it)
                }     
/** 打印
SOURCE 1: 100
SOURCE 1: 1000
SOURCE 2: 900
SOURCE 3: 2000
SOURCE 3: 4000 
*/

不知道什么上面会打印100出来，下面才是我们真正想要

• sample

  Observable.concat(source1, source2, source3)
                .sample(1, TimeUnit.SECONDS)
                .subscribe {
                    println(it)
                }
  /**
SOURCE 1: 900
SOURCE 2: 900
SOURCE 3: 2000
  */

• throttleWithTimeout

 Observable.concat(source1, source2, source3)
                .throttleWithTimeout(1, TimeUnit.SECONDS)
                .subscribe {
                    println(it)
                }
/**
SOURCE 2: 900
SOURCE 3: 2000
SOURCE 3: 4000
*/