image.png

我們在上一篇文章中介紹了 RxSwift 基礎(chǔ)的部分. 現(xiàn)在我們來學(xué)習(xí)一些操作符, 來學(xué)習(xí)一下 FRP 中的F(unctional) 部分

Schedulers

我們先學(xué)習(xí)一下之前就已經(jīng)提到過的, 但是沒有詳細(xì)介紹的 Schedulers

Schedulers 最常見的用法就是告訴 Observables 和訂閱者 應(yīng)該在哪個線程或者隊列中發(fā)送事件,或者通知。

關(guān)于 Schedulers 最常見的操作符是observerOn 和 subscribleOn

通常情況下 Observables 會在它被訂閱的那個線程發(fā)送事件或者通知。

ObserveOn

ObserveOn 指定 Observables 發(fā)送事件的線程或者隊列。它不會改變它執(zhí)行的線程。

舉一個跟 part1 很相似的例子：

let observable = Observable<String>.create { (observer) -> Disposable in
    DispatchQueue.global(qos: .default).async {
        Thread.sleep(forTimeInterval: 10)
        DispatchQueue.main.async {
            observer.onNext("Hello dummy ??")
            observer.onCompleted()
        }
    }
    return Disposables.create()
}

假設(shè)訂閱者是一個 UI 層的東西， 比如說是一個 UIViewController 或者 UIView

DispatchQueue.global(qos: .default).async

我們把這個任務(wù)放在子線程中去執(zhí)行， 以免阻塞 UI

DispatchQueue.main.async{ ...

我們需要在主線程中去更新 UI, 你應(yīng)該知道 UIKit 要求對 UI 的操作都必須在主線程中進(jìn)行。所以這些操作對你來說一定是很熟悉的了。

記下來使用 ObserveOn 來重構(gòu)一下這段代碼

let observable = Observable<String>.create({ (observer) -> Disposable in
    DispatchQueue.global(qos: .default).async {
        Thread.sleep(forTimeInterval: 10)
        observer.onNext("Hello dummy ??")
        observer.onCompleted()
    }
        return Disposables.create()
    }).observeOn(MainScheduler.instance)

我們刪掉了 DispatchQueue.main.async {} 然后添加了 .observeOn(MainScheduler.instance)。 這個就可以讓所有的事件都在主線程中被發(fā)送出去。就是這么簡單。 "Hello dummy ??" 這個元素就能夠很安全的被發(fā)送給 UI 的元素， 因為我們可以很確定他會在主線程中被發(fā)送出去。

observable.subscribe(onNext: { [weak self] (element) in
    self?.label.text = element
}).addDisposableTo(disposeBag)

ObserveOn 大概是最常見的線程調(diào)度操作符了。你希望 Observables 包含了所有的邏輯, 和線程操作, 讓訂閱者盡可能的簡單。所以我們接下來再了解一下 subscribeOn 這個操作符。

SubscribeOn (Optional)

這是一個非常先進(jìn)的操作符。你可以先跳過這部分, 以后再來研究??

subscribeOn 跟 ObserveOn 非常的相似。但是他只能改變 Observable 將要執(zhí)行的任務(wù)所在的線程。

let observable = Observable<String>.create { (observer) -> Disposable in
    Thread.sleep(forTimeInterval: 10)
    observer.onNext("Hello dummy ??")
    observer.onCompleted()
    return Disposables.create()
} 
observable
    .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .default))
    .subscribe(onNext: { [weak self] (element) in
        self?.label.text = element
    }).addDisposableTo(disposeBag)

上面的代碼中, 我刪掉了 Observable 中的 DispatchQueue.global(qos: .default).async {} 是這個訂閱者告訴他應(yīng)該在一個 global queue 中執(zhí)行下面的操作, 以免阻塞 UI. 很明顯這回導(dǎo)致一個異常的拋出, 之前提到過： 這回導(dǎo)致 Observable 在全局隊列中執(zhí)行, 也會在全局隊列中發(fā)出事件。只需要添加在 Observable 中添加 .observeOn(MainScheduler.instance)就能避免這個問題。

let observable = Observable<String>.create { (observer) -> Disposable in
    Thread.sleep(forTimeInterval: 10)
    observer.onNext("Hello dummy ??")
    observer.onCompleted()
        return Disposables.create()
}.observeOn(MainScheduler.instance)      
observable
    .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .default))
    .subscribe(onNext: { [weak self] (element) in
        self?.label.text = element
    }).addDisposableTo(disposeBag)

添加之后，就能夠發(fā)現(xiàn)剛剛說到的問題已經(jīng)解決掉了。

我們什么時候應(yīng)該用 observeOn 呢？最常見的場景是:如果在 Observable 不需要在后臺執(zhí)行耗時操作(讀取數(shù)據(jù), 大的計算任務(wù))的話.我不認(rèn)為這是非常頻繁的事情。但是，come on! 多知道一個你能用的工具 ??不是件很 cool 的事情嗎？

Scheduler Types

做為 RxSwift 菜鳥, 好奇 observeOn 和 MainScheduler.instance 沒什么關(guān)系。你可以自己創(chuàng)建一個線程或者直接使用已經(jīng)創(chuàng)建好了的。如果你很好奇的話這里有很多。 這也沒什么好復(fù)雜的， 就是對 GCD 和 NSOperation 的封裝而已。

Transforming Operators

現(xiàn)在你已經(jīng)知道兩種操作符了: 創(chuàng)建操作符(create、interval、just) 和 功能操作符(observeOn, subscribeOn)。 現(xiàn)在再學(xué)一些轉(zhuǎn)換操作符吧！

Map

這是非常簡單，但非常有用的操作符。它也可能是你未來最常用的一個操作符號。

let observerable = Observable<Int>.create { (observer) -> Disposable in
    observer.onNext(1)
    return Disposables.create()
}        
let boolObservable: Observable<Bool> = observerable.map{(element) -> Bool in
    if element == 0 {
        return false
    }
    return true
}
boolObservable.subscribe(onNext: { (boolElement) in
    print(boolElement)
}).addDisposableTo(disposeBag)

Map 操作符號，改變了序列中值的類型。他映射了一個 Observable 所以他以你告訴他的新的方式發(fā)送事件。在這個例子中, 我們將一個 Int 類型的 Observable 映射成了一個 Bool 類型。

所以這個例子的結(jié)果是

true

Scan

scan 要復(fù)雜一些了。

let observable = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("D")
    observer.onNext("U")
    observer.onNext("M")
    observer.onNext("M")
    observer.onNext("Y")
    return Disposables.create()
}
observable.scan("") { (lastValue, currentValue) -> String in
    return lastValue + currentValue
}.subscribe(onNext: { (element) in
    print(element)
}).addDisposableTo(disposeBag)

在這個例子中會輸出

D
DU
DUM
DUMM
DUMMY

scan操作符, 讓你可以通過上一個值來改變這一個值。他也被稱作元素堆積。上面代碼中的 “”是掃描參數(shù)傳遞的起始值。還是想著能干什么呢？

let observable = Observable<Int>.create { (observer) -> Disposable in
    observer.onNext(1)
    observer.onNext(2)
    observer.onNext(3)
    observer.onNext(4)
    observer.onNext(5)
    return Disposables.create()
}
observable.scan(1) { (lastValue, currentValue) -> Int in
    return lastValue + currentValue
}.subscribe(onNext: { (element) in
    print(element)
}).addDisposableTo(disposeBag)

這是通過 scan 操作符計算 5 的階層。 算出來的答案是: 120

Marin 給了一個更有用的例子 關(guān)于按鈕的 selected 狀態(tài)

let button = UIButton()
button.rx.tap.scan(false) { last, new in
    return !last
}.subscribe(onNext: { (element) in
    print("tap: \(element)")
}).addDisposableTo(disposeBag)

現(xiàn)在你知道他能干什么了吧？ 當(dāng)然還有很多其他的轉(zhuǎn)換操作符。

Filtering Operators

發(fā)出事件是很重要的事情, 但是很多情況下我們還需要過濾掉一些沒用的事件。這就是 filter 操作符所做的事什么。

Filter

決定那些事件是要響應(yīng)的那些是要過濾掉的。

let observerable = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("??")
    observer.onNext("??")
    observer.onNext("??")
    observer.onNext("??")
    observer.onNext("??")
    return Disposables.create()
}
observerable.filter { (element) -> Bool in
    return element == "??"
}.subscribe(onNext: { (element) in
    print(element)
}).addDisposableTo(disposeBag)

輸出

??
??

Debounce

簡單且有用

observerable
    .debounce(2, scheduler: MainScheduler.instance)
    .subscribe(onNext: { (element) in
        print(element)
    }).addDisposableTo(disposeBag)

debounce 會過濾掉2秒以內(nèi)的所有事件, 如果事件a在上一次事件之后的0.5秒被發(fā)送出來。那么他就會被過濾掉。如果他在上次事件的2.5秒被發(fā)送出來。那么他就會被接受到。需要注意的是, 如果就算當(dāng)前時間之后沒有其他的事件，他也要在2秒之后被發(fā)送出來。

譯者: 需要注意的 debounce 和 throttle 的區(qū)別。還有 Obj-C 中的 ReactiveCocoa 中的 throttle 的區(qū)別。

Combining Operator

聯(lián)合操作符讓你可以把多個 Observable 轉(zhuǎn)換成一個。

Merge

合并只是將多個 Observable 發(fā)送的事件合并到一個 Observable 中。

let observable = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("??")
    observer.onNext("??")
    return Disposables.create()
}
let observable2 = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("??")
    observer.onNext("??")
    return Disposables.create()
}
Observable.of(observable, observable2).merge().subscribe(onNext: { (element) in
    print(element)
}).addDisposableTo(disposeBag)     

??
??
??
??

Zip

Zip 將每個 Observable 發(fā)出來的值合并成一個值。

let observable = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("??")
    observer.onNext("??")
    return Disposables.create()
}
let observable2 = Observable<String>.create { (observer) -> Disposable in
    observer.onNext("??")
    observer.onNext("??")
    return Disposables.create()
}
Observable.zip(observable ,observable2).subscribe(onNext: { (element) in
    print(element)
}).addDisposableTo(disposeBag)

????
????

這是一個很有用的操作符。還是舉一個例子吧！ 假如你有兩個網(wǎng)絡(luò)請求, 你需要等到他們都結(jié)束之后再進(jìn)行下一步操作。

let observable = Observable<String>.create { (observer) -> Disposable in
    DispatchQueue.main.async {
        Thread.sleep(forTimeInterval: 3)
        observer.onNext("fetched from sever 1")
    }
    return Disposables.create()
}
let observable2 = Observable<String>.create { (observer) -> Disposable in
    DispatchQueue.main.async {
        Thread.sleep(forTimeInterval: 2)
        observer.onNext("fetched from sever 2")
    }
    return Disposables.create()
}
Observable.zip(observable, observable2)
    .subscribe(onNext: { (element) in
        print(element)
    }).addDisposableTo(disposeBag)

Zip 會等到兩個 Observable 都結(jié)束之后將兩個請求的結(jié)果合并成一個值發(fā)送出來。

Other Operators

還有很多有趣的操作符, 比如 reduce、 takeUntil 等等。我認(rèn)為如果你什么時候有了一些想法, 你也會很容易的找到他們。他們非常的強(qiáng)大, 能讓你快速簡單的操作事件序列。

Mixing Operators

這個教程不需要具體的實例項目, 但是能快的將各種操作符搭配使用。我們來做一個實驗吧：工具根據(jù)事件改變視圖的顏色。

Observable<NSDate>.create { (observer) -> Disposable in
    DispatchQueue.global(qos: .default).async {
        while true {
            Thread.sleep(forTimeInterval: 0.01)
            observer.onNext(NSDate())
        }
    }
    return Disposables.create()
    }// 需要在主線程中刷新 UI
    .observeOn(MainScheduler.instance)
    // 我們只需要能夠被2整除的事件
    .filter { (date) -> Bool in
        return Int(date.timeIntervalSince1970) % 2 == 0
    }
    // 將數(shù)據(jù)轉(zhuǎn)換成顏色
    .map { (date) -> UIColor in
        let interval: Int = Int(date.timeIntervalSince1970)
        let color1 = CGFloat( Double(((interval * 1) % 255)) / 255.0)
        let color2 = CGFloat( Double(((interval * 2) % 255)) / 255.0)
        let color3 = CGFloat( Double(((interval * 3) % 255)) / 255.0)
        return UIColor(red: color1, green: color2, blue: color3, alpha: 1)
    }
    .subscribe(onNext: {[weak self] (color) in
        self?.demoView.backgroundColor = color
    }).addDisposableTo(disposeBag)

You can find more examples in the RxSwfit playgrounds

That's it!

你知道了太多了。剩下的就是 Subjects 了

原文地址