RxSwift - 操作符四

作者: aven_kang | 来源:发表于2022-07-07 17:04 被阅读0次

sample

不定期的对 Observable 取样

smple

sample 操作符将不定期的对源 Observable 进行取样操作。通过第二个 Observable 来控制取样时机。一旦第二个
Observable 发出一个元素，就从源 Observable 中取出最后产生的元素。

        let source = PublishSubject<Int>()
        let notifier = PublishSubject<String>()
        source
        .sample(notifier)
        .subscribe(onNext: { print($0) })
        .disposed(by: bag)

        source.onNext(1)

        //让源序列接收接收消息
        notifier.onNext("A")

        source.onNext(2)

        //让源序列接收接收消息
        notifier.onNext("B")
        notifier.onNext("C")

        source.onNext(3)
        source.onNext(4)

        //让源序列接收接收消息
        notifier.onNext("D")

        source.onNext(5)

        //让源序列接收接收消息
        notifier.onCompleted()

输出结果

分析:从输出结果来看，我们会发现3不见了，为什么不见了呢，我们回到代码notifier.onNext("D")，在这里，发出D之后呢，我们回到source源上，发现最新的是4，并不是3，也就是3会被丢弃了，也就是说，哪怕中间的再多的事件，但是我也是根据监听事件取源的最新一个事件

scan

持续的将 Observable 的每一个元素应用一个函数，然后发出每一次函数返回的结果

scan

scan 操作符将对第一个元素应用一个函数，将结果作为第一个元素发出。然后，将结果作为参数填入到第二个元素的应用函数中，创建第二个元素。以此类推，直到遍历完全部的元素。这种操作符在其他地方有时候被称作是 accumulator 。

        /// scan
       Observable.of(1,2,3,4).scan(0) { a, b in
            return a+b
        }.subscribe { element in
            print(element)
        }.disposed(by: bag)
        // 输出结果
        // next(1)
        // next(3)
        // next(6)
        // next(10)
        // completed

        /// reduce
        Observable.of(1,2,3,4).reduce(0) { a, b in
            return a+b
        }.subscribe { element in
            print(element)
        }.disposed(by: bag)
        // 输出结果
        // next(10)
        // completed

shareReplay

使观察者共享 Observable ，观察者会立即收到最新的元素，即使这些元素是在订阅前产生的
shar eReplay 操作符将使得观察者共享源 Observable ，并且缓存最新的 n 个元素，将这些元素直接发送给新的观察者。

        let seq = PublishSubject<Int>()
           
        let ob = seq.map { (i) -> Int in
            print("map 被调用 ：---\(i)")
            return i * 2
        }

        let _ = ob.subscribe(onNext: { (num) in
            print("--第一次订阅--\(num)")
        }, onError: nil, onCompleted: nil, onDisposed: nil)


        let _ = ob.subscribe(onNext: { (num) in
            print("--第二次订阅--\(num)")
        }, onError: nil, onCompleted: nil, onDisposed: nil)

        seq.onNext(1)

输出结果

map 被调用 ：---1
--第一次订阅--2
map 被调用 ：---1
--第二次订阅--2

这里可以看出来map被调用了两次，我们再看看加了share的情况

 let seq = PublishSubject<Int>()
           
        let ob = seq.map { (i) -> Int in
            print("map 被调用 ：---\(i)")
            return i * 2
        }
            .share(replay: 0, scope: .forever)

        let _ = ob.subscribe(onNext: { (num) in
            print("--第一次订阅--\(num)")
        }, onError: nil, onCompleted: nil, onDisposed: nil)


        let _ = ob.subscribe(onNext: { (num) in
            print("--第二次订阅--\(num)")
        }, onError: nil, onCompleted: nil, onDisposed: nil)

        seq.onNext(1)

输出结果
map 被调用 ：---1
--第一次订阅--2
--第二次订阅--2

总结：
从上面例子可以知道：
发送一次onNext事件，就对应一个Observable，而所有的观察者都只对应这个Observable，Observable共享给所有的观察者
share(replay:scope:)操作符使得观察者共享Observable，并且缓存最新的n个元素，将这些元素直接发送给新的观察者

用例2

无share

        let net = Observable<String>.create { observe in
            
            print("开始网络请求")
            observe.onNext("Response__Result")
            observe.onCompleted()
            return Disposables.create()
            
        }
        
        net.subscribe { element in
            print("第一次订阅\(element)",Thread.current)
        }.disposed(by: bag)
        
        DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 2) {
            net.subscribe { element in
                print("第二次订阅\(element)",Thread.current)
            }.disposed(by: self.bag)
        }

输出结果

开始网络请求
第一次订阅next(Response__Result) <_NSMainThread: 0x6000028dc700>{number = 1, name = main}
第一次订阅completed <_NSMainThread: 0x6000028dc700>{number = 1, name = main}
开始网络请求
第二次订阅next(Response__Result) <_NSMainThread: 0x6000028dc700>{number = 1, name = main}
第二次订阅completed <_NSMainThread: 0x6000028dc700>{number = 1, name = main}

分析:从打印结果来看，没用share的，订阅了两次，就触发了两次next事件

有share

       let net = Observable<String>.create { observe in
           
           print("开始网络请求")
           observe.onNext("Response__Result")
           observe.onCompleted()
           return Disposables.create()
           
       }.share(replay: 0, scope: .forever)
       
       net.subscribe { element in
           print("第一次订阅\(element)",Thread.current)
       }.disposed(by: bag)
       
       DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 2) {
           net.subscribe { element in
               print("第二次订阅\(element)",Thread.current)
           }.disposed(by: self.bag)
       }

输出结果

开始网络请求
第一次订阅next(Response__Result) <_NSMainThread: 0x600001dd4740>{number = 1, name = main}
第一次订阅completed <_NSMainThread: 0x600001dd4740>{number = 1, name = main}
第二次订阅completed <_NSMainThread: 0x600001dd4740>{number = 1, name = main}

分析:从打印结果来看，只执行了一次Next事件，相当于多个观察者共享一个Observable,也只响应这个Observable

single

限制 Observable 只有一个元素，否出发出一个 error事件
single 操作符将限制 Observable 只产生一个元素。如果Observable 只有一个元素，它将镜像这个 Observable 。如果 Observable 没有元素或者元素数量大于一，它将产生一个 error 事件。

       let ob = Observable<String>.create { observe in
            
            observe.onNext("1")
            observe.onNext("2")
            
            return Disposables.create()
        }.single()
            
    
        ob.subscribe { element in
            print(element)
        }.disposed(by: bag)
            
        ob.subscribe { element in
            print(element)
        }.disposed(by: bag)

输出结果

next(1)
error(Sequence contains more than one element.)
next(1)
error(Sequence contains more than one element.)

分析:从上述代码可以看出来，我们发出了两个onNext事件，但是后面的订阅者中可以看出打印结果，第一次是正常发出事件，当第二次还存在事件的时候，就强制发出了error，结束了该序列

skip

跳过 Observable 中头 n 个元素

            Observable.of(1,2,3,4,5,6,7)
            .skip(2)
            .subscribe { element in
                switch element {
                case .next(let index):
                    print(index)
                default:
                    break
                }
            }
            .disposed(by: bag)

输出结果

skipUntil

        let sourceSequence = PublishSubject<String>()
        let referenceSequence = PublishSubject<String>()
        
        sourceSequence.skip(until: referenceSequence)
            .subscribe { element in
                print(element)
            }.disposed(by: bag)
        
        sourceSequence.onNext("1")
        sourceSequence.onNext("2")
        sourceSequence.onNext("3")
        referenceSequence.onNext("4")
//
        sourceSequence.onNext("5")
        sourceSequence.onNext("6")
        sourceSequence.onNext("7")

输出结果

next(5)
next(6)
next(7)

总结:这里有一个注意点，如果你不执行referenceSequence.onNext("4")，那么sourceSequence所有的onNext事件将都不会响应，也就是不会有任何输出结果

skipWhile

            Observable<Int>.of(1,2,5,3,2,4,3,2,1)
            .skip { element in
                return element < 4
            }
            .subscribe { element in
                print(element)
            }
            .disposed(by: bag)

输出结果

next(5)
next(3)
next(2)
next(4)
next(3)
next(2)
next(1)
completed

分析:程序中，我们设定了element < 4的条件，第一个元素是1，小于4，成立，第二个元素是2，小于4程序，第三个是5，并不小于4，返回false，而skipWhile的功能就是，直到遇到第一个false的条件，那么前面所有的元素都将被丢弃，从打印结果来看，开头的1，2是没有打印的，便可印证

startWith

将一些元素插入到序列的头部
startWith操作符会在 Observable 头部插入一些元素。（如果你想在尾部加入一些元素可以用concat）

        Observable<Int>.of(1,2,3,4)
            .startWith(6,5)
            .startWith(7)
            .startWith(8)
            .subscribe { element in
                print(element)
            }
            .disposed(by: bag)

输出结果

next(8)
next(7)
next(6)
next(5)
next(1)
next(2)
next(3)
next(4)
completed

subscribeOn

指定 Observable 在那个 Scheduler 执行
ReactiveX 使用 Scheduler 来让 Observable 支持多线程。你可以使用 subscribeOn操作符，来指示 Observable 在哪个Scheduler 执行。
observeOn操作符非常相似。它指示 Observable 在哪个Scheduler 发出通知。

 PublishSubject<Int>.of(1,2,3).subscribe(on: MainScheduler.instance)
            .subscribe { element in
                print(Thread.current)
                print(element)
            }.disposed(by: bag)

输出结果

<_NSMainThread: 0x6000030a4700>{number = 1, name = main}
next(1)
<_NSMainThread: 0x6000030a4700>{number = 1, name = main}
next(2)
<_NSMainThread: 0x6000030a4700>{number = 1, name = main}
next(3)
<_NSMainThread: 0x6000030a4700>{number = 1, name = main}
completed

take

仅仅从 Observable 中发出头 n 个元素
通过 take 操作符你可以只发出头 n 个元素。并且忽略掉后面的元素，直接结束序列。

        PublishSubject<Int>.of(1,2,3,4,5).subscribe(on: MainScheduler.instance)
            .take(2)
            .subscribe { element in
                print(Thread.current)
                print(element)
            }.disposed(by: bag)

输出结果

<_NSMainThread: 0x6000027d0900>{number = 1, name = main}
next(1)
<_NSMainThread: 0x6000027d0900>{number = 1, name = main}
next(2)
<_NSMainThread: 0x6000027d0900>{number = 1, name = main}
completed

takeLast

仅仅从 Observable 中发出尾部 n 个元素
通过 takeLast 操作符你可以只发出尾部 n 个元素。并且忽略掉前面的元素。

 PublishSubject<Int>.of(1,2,3,4,5).subscribe(on: MainScheduler.instance)
            .takeLast(2)
            .subscribe { element in
                print(Thread.current)
                print(element)
            }.disposed(by: bag)

输出结果

<_NSMainThread: 0x6000025dc440>{number = 1, name = main}
next(4)
<_NSMainThread: 0x6000025dc440>{number = 1, name = main}
next(5)
<_NSMainThread: 0x6000025dc440>{number = 1, name = main}
completed

takeUntil

忽略掉在第二个 Observable 产生事件后发出的那部分元素
takeUntil 操作符将镜像源 Observable ，它同时观测第二个
Observable 。一旦第二个 Observable 发出一个元素或者产生一个终止事件，那个镜像的 Observable 将立即终止。

        let soucecSequence = PublishSubject<Int>()
        let referenceSequence = PublishSubject<Int>()
        
        soucecSequence.take(until: referenceSequence)
            .subscribe { element in
                print(element)
            }
        
        soucecSequence.onNext(1)
        soucecSequence.onNext(2)
        soucecSequence.onNext(3)
        
        referenceSequence.onNext(4)
        
        soucecSequence.onNext(5)
        soucecSequence.onNext(6)
        soucecSequence.onNext(7)

输出结果

next(1)
next(2)
next(3)
completed

takeWhile

镜像一个 Observable 直到某个元素的判定为 false
takeWhile 操作符将镜像源 Observable 直到某个元素的判定为 false。此时，这个镜像的 Observable 将立即终止。

        PublishSubject<Int>.of(1,2,3,4,5,6,7)
            .take(while: { index in
                return index < 4
            })
            .subscribe {
                print($0)
            }.disposed(by: bag)

输出结果

next(1)
next(2)
next(3)
completed

timeout

设置超时时间，超过规定时间的事件将发送 error

        let pub = PublishSubject<Int>()
        pub.timeout(RxTimeInterval.seconds(3), scheduler: MainScheduler.instance).subscribe { element in
            print(element)
        }.disposed(by: bag)
        
        DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 4) {
            pub.onNext(3)
        }

timer

创建一个 Observable 在一段延时后，产生唯一的一个元素
timer 操作符将创建一个 Observable ，它在经过设定的一段时间后，产生唯一的一个元素。

public static func timer( _ dueTime: RxTimeInterval, // 初始延时
period: RxTimeInterval?, // 时间间隔
scheduler: SchedulerType
) -> Observable<E>

 PublishSubject<Int>
            .timer(RxTimeInterval.seconds(2), period: RxTimeInterval.seconds(3), scheduler: MainScheduler.instance)
            .subscribe { element in
            print(element)
        }.disposed(by: bag)

RxSwift - 操作符四

sample

scan

shareReplay

用例2

single

skip

skipUntil

skipWhile

startWith

subscribeOn

take

takeLast

takeUntil

takeWhile

timeout

timer

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读