前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
首页
学习
活动
专区
工具
TVP最新优惠活动
返回腾讯云官网
社区首页 >专栏 >RxJava 过滤操作符

RxJava 过滤操作符

作者头像
三流编程
发布于 2018-09-11 15:54:48
1.6K0
发布于 2018-09-11 15:54:48
举报
文章被收录于专栏：三流程序员的挣扎三流程序员的挣扎

blockingFirst/blockingLast

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

代码语言：javascript
复制
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)}")

结果

代码语言：javascript
复制
first 1
first default 10
last 4
last default 4

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

blockingSingle

代码语言：javascript
复制
val ob = Observable.just(1)
val i = ob.blockingSingle()
Log.e("RX", "$i")

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

distinct

代码语言：javascript
复制
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)

去掉发射的重复数据。

代码语言：javascript
复制
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

代码语言：javascript
复制
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

代码语言：javascript
复制
// 收到 3,3,3,6,6,6,9
ob.distinct ({ it }, {MyCollection<Int>()}).subscribe(observerInt)

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

代码语言：javascript
复制
// 只收到 onComplete
ob.distinct ({ it % 3 }, {MyCollection<Int>()}).subscribe(observerInt)

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

distinctUntilChanged

代码语言：javascript
复制
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 来判断两个数据是否相等。

代码语言：javascript
复制
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。

代码语言：javascript
复制
// 越界，则发射 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()
代码语言：javascript
复制
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

过滤。

代码语言：javascript
复制
Observable.just(1,2,3,4,5,6)
     .filter({ it % 2 == 0 }).subscribe(observerInt)

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

代码语言：javascript
复制
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 类型的数据。

代码语言：javascript
复制
// 只收到 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 事件。

代码语言：javascript
复制
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

代码语言：javascript
复制
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

代码语言：javascript
复制
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

代码语言：javascript
复制
// 跳过指定数目
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)
代码语言：javascript
复制
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

代码语言：javascript
复制
// 最多发多少个数据
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)
代码语言：javascript
复制
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

代码语言：javascript
复制
public final Observable<T> debounce(long timeout, TimeUnit unit)

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

代码语言：javascript
复制
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.

代码语言：javascript
复制
public final <U> Observable<T> debounce(Function<? super T, ? extends ObservableSource<U>> debounceSelector)
代码语言：javascript
复制
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 产生的结果，在指定的间隔周期内进行采样。

代码语言：javascript
复制
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)
代码语言：javascript
复制
// 收到 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") }
代码语言：javascript
复制
public final <U> Observable<T> sample(ObservableSource<U> sampler)
public final <U> Observable<T> sample(ObservableSource<U> sampler, boolean emitLast)

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

代码语言：javascript
复制
// 收到 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 方法，两者没有任何区别，只是它的重载方法只有两个。

代码语言：javascript
复制
public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit)
public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit, Scheduler scheduler)

throttleFirst

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

代码语言：javascript
复制
// 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") }
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2018.06.09 ，如有侵权请联系 cloudcommunity@tencent.com 删除
其他

本文分享自 作者个人站点/博客 前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ，欢迎热爱写作的你一起参与！

其他
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
LV.
文章
0
获赞
0
目录
  • blockingFirst/blockingLast
  • blockingSingle
  • distinct
  • distinctUntilChanged
  • elementAt/elementAtOrError/firstElement/first/firstOrError/lastElement/last/lastOrError
  • filter
  • ofType
  • ignoreElements
  • singleElement
  • single/singleOrError
  • skip/skipLast/skipUntil/skipWhile
  • take/takeLast/takeUntil/takeWhile
  • debounce/throttleWithTimeout
  • sample/throttleLast
  • throttleFirst
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2024 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论