美文网首页
RxJava 过滤操作符

RxJava 过滤操作符

作者: 三流之路 | 来源:发表于2018-06-09 22:01 被阅读0次

    ReactiveX 系列文章目录

    blockingFirst/blockingLast

    阻塞直到返回第一个/最后一个才发射数据。

    Log.e("RX", "first ${Observable.just(1,2,3,4).blockingFirst()}")
// Observable 没发射，返回默认值
Log.e("RX", "first default ${Observable.empty<Int>().blockingFirst(10)}")
Log.e("RX", "last ${Observable.just(1,2,3,4).blockingLast()}")
Log.e("RX", "last default ${Observable.empty<Int>().blockingLast(10)}")

    结果

    first 1
first default 10
last 4
last default 4

    现在想到的应用场景也许是某些 Callable 返回的数据之类，先过滤拿到第一个，然后再用 Observable 发射这第一个数。

    blockingSingle

    val ob = Observable.just(1)
val i = ob.blockingSingle()
Log.e("RX", "$i")

    如果发射数据只有一个，返回这个发射的值；如果多于一个，抛出异常；如果没有发射数据，返回参数传递的默认值。

    distinct

    public final Observable<T> distinct()
public final <K> Observable<T> distinct(Function<? super T, K> keySelector)
public final <K> Observable<T> distinct(Function<? super T, K> keySelector, Callable<? extends Collection<? super K>> collectionSupplier)

    去掉发射的重复数据。

    val ob = Observable.fromArray(1,2,3,2,2,3,3,4,5,6,6,6,9)

// 默认规则去重，收到 1,2,3,4,5,6,9
ob.distinct().subscribe(observerInt)

// 求余 3 后比较，4 认为和 1 重复，5 认为和 2 重复，最终只收到 1,2,3
ob.distinct { it % 3 }.subscribe(observerInt)

    最后一个构造方法，第三个参数提供一个集合，看源码是先通过前面的 keySelector 方法获取转换后的数据，然后往集合里面 add 要转换后的数据，只有这个 add 返回 true，才会通过 onNext 发射数据。而没有这个参数的方法，内部都是用的 HashSet，不许添加重复元素，这样前面的规则判断出哪些是重复，然后这里添加。

    现在我写一个没什么意义的 Collection

    class MyCollection<Int>: ArrayList<Int>() {
    override fun add(element: Int): Boolean {
        super.add(element)
        return (element == 6 || element == 9 || element == 3)
    }
}

    就是说只要这个值是 3,6,9 中的一个，add 方法就返回 true，其余值一律 false

    // 收到 3,3,3,6,6,6,9
ob.distinct ({ it }, {MyCollection<Int>()}).subscribe(observerInt)

    虽然根据默认规则，多个 6 是相同的，但是往集合 add 返回值是 true，所以仍然发射出来，所以有多个 6。

    // 只收到 onComplete
ob.distinct ({ it % 3 }, {MyCollection<Int>()}).subscribe(observerInt)

    因为规则是 it % 3，所以大于 3 的数经转换全部变成了 0,1,2，然后 add 返回值全部 false，所以一个都没有发射出来。

    distinctUntilChanged

    public final Observable<T> distinctUntilChanged()
public final <K> Observable<T> distinctUntilChanged(Function<? super T, K> keySelector)
public final Observable<T> distinctUntilChanged(BiPredicate<? super T, ? super T> comparer)

    去除相邻的重复数据。也是通过一个 keySelector 来对数据做相应转换，然后通过 BiPredicate 对象 compare 来判断两个数据是否相等。

    val ob = Observable.fromArray(1,2,3,2,2,3,3,4,5,6,6,6,9)
// 收到 1,2,3,2,3,4,5,6,9
//            ob.distinctUntilChanged().subscribe(observerInt)
// 都先转成了 1，然后用默认的 equals 判断，所有都相等，所以只收到 1
//            ob.distinctUntilChanged( Function<Int, Int> { 1 } ).subscribe(observerInt)
// 虽然数据没转换，但判断是否相等时都认为是一样的，所以也只收到 1
ob.distinctUntilChanged({ _, _ -> true}).subscribe(observerInt)

    elementAt/elementAtOrError/firstElement/first/firstOrError/lastElement/last/lastOrError

    只发射指定位置的数据。返回 Maybe。

    // 越界，则发射 onComplete，否则发射指定 index 的数据
public final Maybe<T> elementAt(long index)
// 越界，用默认值 defaultItem
public final Single<T> elementAt(long index, T defaultItem)
// 越界，抛出异常
public final Single<T> elementAtOrError(long index)

public final Maybe<T> firstElement() {
    return elementAt(0L);
}
public final Single<T> first(T defaultItem) {
    return elementAt(0L, defaultItem);
}
public final Single<T> firstOrError() {
    return elementAtOrError(0L);
}

// 空，发射 onComplete
public final Maybe<T> lastElement()
// 空，用默认值
public final Single<T> last(T defaultItem)
// 空，抛出异常
public final Single<T> lastOrError()

    Observable.just(1, 2, 3, 4).elementAt(2)
            .subscribe({ textView.text = "${textView.text}\n $it" })

Observable.just(1, 2, 3, 4).elementAt(6, 0)
                  .subscribe(Consumer<Int> { textView.text = "${textView.text}\n $it" })

Observable.just(1, 2, 3, 4).elementAtOrError(6)
                   .subscribe(Consumer<Int> { textView.text = "${textView.text}\n $it" })

    filter

    过滤。

    Observable.just(1,2,3,4,5,6)
     .filter({ it % 2 == 0 }).subscribe(observerInt)

    filter 的参数是 Predicate，test 方法用于过滤，返回 false 的丢弃。

    public interface Predicate<T> {
    /**
     * Test the given input value and return a boolean.
     * @param t the value
     * @return the boolean result
     * @throws Exception on error
     */
    boolean test(@NonNull T t) throws Exception;
}

    ofType

    内部调用 filter，过滤指定的 Class 类型的数据。

    // 只收到 a 和 b
Observable.just(1,2,"a","b", 4L, true)
                                .ofType(String::class.java).subscribe { t-> textView.text = "${textView.text}\n $t"}

    ignoreElements

    忽略所有数据，只接受 complete 或 error 事件。

    Observable.range(0, 5)
        .ignoreElements().subscribe(object : Action {
            override fun run() {
                textView.text = "${textView.text}\n complete"
            }
        }, object : Consumer<Throwable> {
            override fun accept(t: Throwable?) {
                textView.text = "${textView.text}\n error"
            }
        })

    singleElement

    val observer = object : MaybeObserver<Int> {
        override fun onSuccess(t: Int) { Log.e("RX", "onSuccess $t") }
        override fun onComplete() { Log.e("RX", "onComplete") }
        override fun onSubscribe(d: Disposable) {}
        override fun onError(e: Throwable) {}
}

// 因为发射了两个数，所以什么都收不到
// val observable = Observable.just(1,2)
// onSuccess 1
// val observable = Observable.just(1)
// onComplete
val observable = Observable.empty<Int>()

observable.singleElement().subscribe(observer)

    single/singleOrError

    val observer = object : SingleObserver<Int> {
    override fun onSuccess(t: Int) { Log.e("RX", "onSuccess $t") }
    override fun onSubscribe(d: Disposable) {}
    override fun onError(e: Throwable) {Log.e("RX", "onError ${e.message}")}
}
// 超过一个，抛异常，进 onError
// val observable = Observable.just(1,2)
// onSuccess 1
// val observable = Observable.just(1)
// 发射数据为空，使用默认值，onSuccess 100
val observable = Observable.empty<Int>()
observable.single(100).subscribe(observer)

// 如果只发射一个值，进入 onSuccess，否则都是 onError
observable.singleOrError().subscribe(observer)

    skip/skipLast/skipUntil/skipWhile

    // 跳过指定数目
public final Observable<T> skip(long count)
// 跳过指定时间
public final Observable<T> skip(long time, TimeUnit unit)
public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler)

// 反方向跳过
public final Observable<T> skipLast(int count)
public final Observable<T> skipLast(long time, TimeUnit unit)
public final Observable<T> skipLast(long time, TimeUnit unit, boolean delayError)
public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler)
public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError)
public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize)

public final <U> Observable<T> skipUntil(ObservableSource<U> other)

public final Observable<T> skipWhile(Predicate<? super T> predicate)

    val ob = Observable.intervalRange(0, 10, 0, 100, TimeUnit.MILLISECONDS)
val consumer  = Consumer<Long>{ Log.e("RX", "$it") }

// 收到 3-9
ob.skip(3).subscribe(consumer)
// 跳过 400ms，收到 4-9
ob.skip(400, TimeUnit.MILLISECONDS).subscribe(consumer)

// 收到 0-6
ob.skipLast(3).subscribe(consumer)

// 直到返回的 Observable 开始发射，这之前源 Observable 发射的数据跳过
// 收到 5-9
ob.skipUntil(Observable.timer(500, TimeUnit.MILLISECONDS))
            .subscribe(consumer)

// 只要条件满足就跳过，收到 4-9
var i = 0
ob.skipWhile { i++ < 4 }.subscribe(consumer)

    take/takeLast/takeUntil/takeWhile

    // 最多发多少个数据
public final Observable<T> take(long count)
// 只发射某个时间间隔前的数据
public final Observable<T> take(long time, TimeUnit unit)
public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler)

// 反方向
public final Observable<T> takeLast(int count)
// 只发射在源 Observable 的 complete 之前某个间隔的数据
public final Observable<T> takeLast(long time, TimeUnit unit)
public final Observable<T> takeLast(long time, TimeUnit unit, boolean delayError)
public final Observable<T> takeLast(long time, TimeUnit unit, Scheduler scheduler)
public final Observable<T> takeLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError)
public final Observable<T> takeLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize)

// 同时用时间和数目的限制
public final Observable<T> takeLast(long count, long time, TimeUnit unit)
public final Observable<T> takeLast(long count, long time, TimeUnit unit, Scheduler scheduler)
public final Observable<T> takeLast(long count, long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize)

// 当 other 发射时不再从源 Observable 中取
public final <U> Observable<T> takeUntil(ObservableSource<U> other)
// 返回 true 时停止 take，发射的数据作为参数
public final Observable<T> takeUntil(Predicate<? super T> stopPredicate)
// 条件满足时一直 take
public final Observable<T> takeWhile(Predicate<? super T> predicate)

    val ob = Observable.intervalRange(0, 10, 0, 100, TimeUnit.MILLISECONDS)
val consumer  = Consumer<Long>{ Log.e("RX", "$it") }

ob.take(3).subscribe(consumer) // 0,1,2

// complete 前 300ms 的数据，即 7,8,9
ob.takeLast(300, TimeUnit.MILLISECONDS).subscribe(consumer)

// 参数的 Observable 开始发射时停止 take，收到 0,1,2
ob.takeUntil(Observable.timer(300, TimeUnit.MILLISECONDS)).subscribe(consumer)

// 发射的数据大于 3 时，停止 take，收到 0,1,2,3,4
ob.takeUntil { it > 3 }.subscribe(consumer)

// 发射的数据小于 3 时就一直 take，收到 0,1,2
ob.takeWhile { it < 3 }.subscribe(consumer)

    debounce/throttleWithTimeout

    public final Observable<T> debounce(long timeout, TimeUnit unit)

    每产生一个数据后，如果在规定的间隔时间内没有别的数据产生，就会发射这个数据，否则忽略该数据。

    Observable.create(ObservableOnSubscribe<Int> { emitter ->
    emitter.onNext(1)
    Thread.sleep(100)
    emitter.onNext(2)
    Thread.sleep(300)
    emitter.onNext(3)
    Thread.sleep(200)
    emitter.onNext(4)
    Thread.sleep(400)
}).debounce(250, TimeUnit.MILLISECONDS)
    .subscribe({
        Log.e("RX", "$it")
    })

    日志显示 2 和 4，首先发射 1，100ms 后发射 2，也就是说发射 1 后的 250ms 以内又发射了其它数，那么忽略 1，2 发射后 250ms 内并没有新数据发出，因为 300ms 后才发出 3，所以接收了 2，3 发射后 200ms 发射 4，那么忽略 3，4 之后没数据再发射，所以接收 4.

    public final <U> Observable<T> debounce(Function<? super T, ? extends ObservableSource<U>> debounceSelector)

    Observable.create(ObservableOnSubscribe<Int> { emitter ->
    emitter.onNext(1)
    Thread.sleep(100)
    emitter.onNext(2)
    Thread.sleep(300)
    emitter.onNext(3)
    Thread.sleep(200)
    emitter.onNext(4)
}).debounce({
    if (it == 2)
        Observable.timer(200, TimeUnit.MILLISECONDS)
    else
        Observable.timer(100, TimeUnit.MILLISECONDS)
  })
  .subscribe({
      Log.e("RX", "$it")
  })

    Function 里方法返回的 Observeable，在它发射结束之前，原始 Observable 发射了新的数据，旧的被忽略。比如发射 1 时，返回的这个 Observable 需要 200ms 后才结束，而 100ms 后就发射了 2，所以 1 没了，对于 2，返回的 Observable 也要 200ms 结束，它结束后再过 100ms 才发了 3，所以 2，3 都在。

    throttleWithTimeout 内部就是直接调用了 debounce。

    sample/throttleLast

    定期扫描源 Observable 产生的结果，在指定的间隔周期内进行采样。

    public final Observable<T> sample(long period, TimeUnit unit)
// emitLast 是否发射最后一个数据，若没有这个参数，默认是 false
public final Observable<T> sample(long period, TimeUnit unit, boolean emitLast)
public final Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler)
public final Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler, boolean emitLast)

    // 收到 3,6,9,12,15,18
Observable.intervalRange(1, 20, 0,100, TimeUnit.MILLISECONDS)
                         .sample(300, TimeUnit.MILLISECONDS).subscribe { Log.e("RX", "$it") }

// 收到 3,6,9,12,15,18,20
Observable.intervalRange(1, 20, 0,100, TimeUnit.MILLISECONDS)
                         .sample(300, TimeUnit.MILLISECONDS, true).subscribe { Log.e("RX", "$it") }

    public final <U> Observable<T> sample(ObservableSource<U> sampler)
public final <U> Observable<T> sample(ObservableSource<U> sampler, boolean emitLast)

    参数里的 Observable 发射时，从源 Observable 中取出一个发射。

    // 收到 1,2,3,...,8,9
Observable.intervalRange(1, 20, 0, 100, TimeUnit.MILLISECONDS)
                .sample(Observable.intervalRange(1, 10, 0, 100, TimeUnit.MILLISECONDS))
                .subscribe { Log.e("RX", "$it") }

// 收到 2,4,6,...,16,18
Observable.intervalRange(1, 20, 0, 100, TimeUnit.MILLISECONDS)
                  .sample(Observable.intervalRange(1, 10, 200, 200, TimeUnit.MILLISECONDS))
                  .subscribe { Log.e("RX", "$it") }

// 什么都收不到
Observable.intervalRange(1, 20, 0, 100, TimeUnit.MILLISECONDS)
        .sample(Observable.just("a","b"))
        .subscribe { Log.e("RX", "$it") }

    throttleLast 内部直接就是调用 sample 方法，两者没有任何区别，只是它的重载方法只有两个。

    public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit)
public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit, Scheduler scheduler)

    throttleFirst

    和 throttleLast 基本一样，throttleLast 是每次采取取最后那个，而 throttleFirst 是第一项。

    // throttleLast 收到 3,6,9,12,15,18
// throttleFirst 收到 1,5,9,13,17
Observable.intervalRange(1, 20, 0,100, TimeUnit.MILLISECONDS)
                    .throttleFirst(300, TimeUnit.MILLISECONDS)
                    .subscribe { Log.e("RX", "$it") }

    相关文章

      网友评论

          本文标题：RxJava 过滤操作符

          本文链接：https://www.haomeiwen.com/subject/xdyxeftx.html

          延伸阅读

          深度阅读

          • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

          栏目导航

          热点阅读

          关于我们|服务条款|联系我们|RxJava 过滤操作符|投稿指南|网站地图|RSS订阅|排版工具|手机版

          提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

          Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

          备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

          本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

          所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！