一、前提：函數(shù)響應(yīng)式編程思想
簡(jiǎn)單來(lái)說(shuō) 函數(shù)響應(yīng)式編程 = 函數(shù)式編程 + 響應(yīng)式編程
（1）函數(shù)式
函數(shù)式編程是種編程方式，它將電腦運(yùn)算視為函數(shù)的計(jì)算。函數(shù)編程語(yǔ)言最重要的基礎(chǔ)是λ演算（lambda calculus），而且λ演算的函數(shù)可以接受函數(shù)當(dāng)作輸入（參數(shù)）和輸出（返回值）。
y = f(x)
x: 參數(shù) f: 函數(shù) y: 返回值
//把 x 作為一個(gè)函數(shù)傳入 --> x = f(x) 例如 x 值是2，2 = 1+1 = 0 + 2
y = f(f(x))//增加了靈活性
let array = [1,2,3,4,5,6,7]
//篩選數(shù)組(1.獲取 > 3的數(shù)字；2. 獲取到的數(shù)字 + 1；3. 獲取偶數(shù)； 4.)
//正常操作--可讀性底，清晰度差
for num in array{
if num > 3{
let number = num + 1
if (number % 2 == 0){
print(number)
}
}
}
//RxSwift函數(shù)式
array.filter{ $0 > 3}//拿到第一個(gè)數(shù)大于3
.filter{($0 + 1) % 2 == 0}// +1 獲取偶數(shù)
.forEach{ print($0)}//輸出
//如果需求變化，函數(shù)式則直接注釋某項(xiàng)filter,完全不影響，正常操作中則需要大改動(dòng)
（2）響應(yīng)式
對(duì)象對(duì)某一數(shù)據(jù)流變化做出響應(yīng)的這種編碼方式
//對(duì)象A和對(duì)象B，A和B有一種“說(shuō)不清”的關(guān)系，A要時(shí)刻監(jiān)控B的行為，對(duì)B的變化也做出相應(yīng)的變化。
例如，A是10歲 ，B比A大10歲，20年后，A是30歲， B則是40歲 。---->類似于KVO
二、RxSwift 簡(jiǎn)單使用
創(chuàng)建序列 -> 訂閱序列 -> 發(fā)送信號(hào) -> 信號(hào)接收
（1）Button點(diǎn)擊事件
//修改事件類型 button.rx.controlEvent(.touchUpOutside)
button.rx.tap.subscribe(onNext: { () in
print("被點(diǎn)擊了")
//處理事件 ……
}).disposed(by: disposeBag)
（2）手勢(shì)點(diǎn)擊事件
let tap = UITapGestureRecognizer.init()
self.view.addGestureRecognizer(tap)
tap.rx.event.subscribe(onNext: { (tap) in
print(tap.view as Any)
//處理事件
}).disposed(by: disposeBag)
（3）輸入框監(jiān)聽(tīng)
//監(jiān)聽(tīng)輸入變化
textfield.rx.text.orEmpty.changed.subscribe(onNext: { (text) in
print(text)
//處理事件 ……
}).disposed(by: disposeBag)
//綁定，數(shù)據(jù)傳遞
textfield.rx.text.bind(to: label.rx.text)
（4）KVO事件監(jiān)聽(tīng)
self.person.rx.observeWeakly(String.self, "name").subscribe(onNext: { (value) in
print(value as Any)
//處理事件 ……
}).disposed(by: disposeBag)
（5）滑動(dòng)事件監(jiān)聽(tīng)
scrollview.rx.contentOffset.subscribe(onNext: { [weak self](content) in
let y = content.y
print(content.y)
//處理事件 ……
}).disposed(by: disposeBag)
（6）定時(shí)器
let timer = Observable<Int>
.interval(1, scheduler: MainScheduler.instance)
timer.subscribe(onNext: { (time) in
print(time)
//處理事件 ……
}).disposed(by: disposeBag)
（7）通知事件
NotificationCenter.default.rx
.notification(UIResponder.keyboardWillShowNotification)
.subscribe(onNext: { (notification) in
//獲取值
let during = notification.userInfo?["UIKeyboardAnimationDurationUserInfoKey"] as? Float
print(during!)
//處理事件 ……
}).disposed(by: disposeBag)
（8）網(wǎng)絡(luò)請(qǐng)求
let url = URL.init(string: "https://www.baidu.com")
URLSession.shared.dataTask(with: url!) { (data, response, error) in
print("network")
print(String.init(data: data!, encoding: .utf8))
}.resume()
URLSession.shared.rx.response(request: URLRequest.init(url: url!))
.subscribe(onNext: { (data) in
print(data)
}, onError: { (error) in
print(error)
}, onCompleted: {
print("請(qǐng)求完成")
}).disposed(by: disposeBag)
三、RxSwift核心邏輯
（1）邏輯流程：創(chuàng)建序列 -->訂閱序列-->發(fā)送信號(hào)-->接受信號(hào)
//1、創(chuàng)建序列
let ob = Observable<Any>.create { (obserber) -> Disposable in
// 3:發(fā)送信號(hào)
obserber.onNext("發(fā)送信號(hào)")
obserber.onCompleted()
// obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
return Disposables.create()
}
//2、訂閱序列
let _ = ob.subscribe(onNext: { (text) in//5、返回銷毀者(_)
//4、接受信號(hào)
print("訂閱到:\(text)")
}, onError: { (error) in
//4、接受信號(hào)
print("error: \(error)")
}, onCompleted: {
//4、接受信號(hào)
print("完成")
}) {
print("銷毀")
}
（2）源碼分析
1.創(chuàng)建序列之Create
Create源碼
//創(chuàng)建返回一個(gè)匿名內(nèi)部類--->AnonymousObservable
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
AnonymousObservable內(nèi)部類源碼
//繼承Producer，初始化并保存了一個(gè)_subscribeHandler的回調(diào)記錄
final private class AnonymousObservable<Element>: Producer<Element> {
// 閉包
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler//保存閉包-->函數(shù)式編程思想
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
通過(guò)源碼分析Observable<Any>.create這一步主要做了:
^創(chuàng)建一個(gè)內(nèi)部類（AnonymousObservable -->繼承Producer）
^保存閉包（_subscribeHandler回調(diào)）
2.訂閱序列之subscribe(onNext
subscribe(onNext源碼
/**
創(chuàng)建一個(gè)觀察者-->AnonymousObserver-->把所有event事件保存到observer里面
返回一個(gè)銷毀者-->Disposables
*/
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
let disposable: Disposable
if let disposed = onDisposed {
disposable = Disposables.create(with: disposed)
}
else {
disposable = Disposables.create()
}
#if DEBUG
let synchronizationTracker = SynchronizationTracker()
#endif
let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
//創(chuàng)建AnonymousObserver觀察者，并且把所有event事件保存到observer里面
let observer = AnonymousObserver<E> { event in
#if DEBUG
synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { synchronizationTracker.unregister() }
#endif
switch event {//判斷當(dāng)前event
case .next(let value)://調(diào)用event.next信息，則會(huì)調(diào)用onNext()
onNext?(value)
case .error(let error):
if let onError = onError {
onError(error)
}
else {
Hooks.defaultErrorHandler(callStack, error)
}
disposable.dispose()
case .completed:
onCompleted?()
disposable.dispose()
}
}
return Disposables.create(//創(chuàng)建一個(gè)銷毀者返回給subscribe
// self.asObservable()---> 就是創(chuàng)建序列的ob ，subscribe --->調(diào)用父類Producer的subscribe
self.asObservable().subscribe(observer),
disposable
)
}
AnonymousObserver觀察者源碼
/**
保存回調(diào)事件閉包（_eventHandler -->所有onNext事件&onError事件&onCompleted事件）
*/
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
typealias Element = ElementType
typealias EventHandler = (Event<Element>) -> Void
private let _eventHandler : EventHandler
init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self._eventHandler = eventHandler
}
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
#if TRACE_RESOURCES
deinit {
_ = Resources.decrementTotal()
}
#endif
}
//進(jìn)入Producer中的subscribe方法
class Producer<Element> : Observable<Element> {
override init() {
super.init()
}
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
if !CurrentThreadScheduler.isScheduleRequired {
// The returned disposable needs to release all references once it was disposed.
let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)//調(diào)用AnonymousObservable的run方法
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
else {
return CurrentThreadScheduler.instance.schedule(()) { _ in
let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
}
}
func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
rxAbstractMethod()
}
}
//進(jìn)入AnonymousObservable中的run方法
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)//調(diào)用了AnonymousObservableSink中的run方法
return (sink: sink, subscription: subscription)
}
}
//進(jìn)入AnonymousObservableSink中調(diào)用run方法
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
typealias E = O.E
typealias Parent = AnonymousObservable<E>
// state
private let _isStopped = AtomicInt(0)
#if DEBUG
fileprivate let _synchronizationTracker = SynchronizationTracker()
#endif
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
func on(_ event: Event<E>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
switch event {
case .next:
if load(self._isStopped) == 1 {
return
}
self.forwardOn(event)
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.forwardOn(event)
self.dispose()
}
}
}
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))//run 方法中的_subscribeHandler，是不是很熟悉，是不是在《創(chuàng)建序列》創(chuàng)建的_subscribeHandler 閉包， 調(diào)用了_subscribeHandler，傳入了AnyObserver(self)-->observer
}
}
是不是有點(diǎn)蒙，來(lái)來(lái)來(lái)， 咱們總結(jié)下:
1.創(chuàng)建一個(gè)觀察者
-->AnonymousObserver（let observer = AnonymousObserver<E> { event in）??保存閉包回調(diào)_eventHandler
-->把所有event事件保存到observer里面
2.返回一個(gè)銷毀者
-->Disposables (return Disposables.create)
-->self.asObservable().subscribe(observer), self.asObservable==創(chuàng)建序列的ob ，subscribe --->調(diào)用父類Producer的subscribe
-->進(jìn)入父類Producer的subscribe方法調(diào)用self.run(observer, cancel: disposer)
-->進(jìn)入AnonymousObservable中的run方法調(diào)用sink.run(self)
-->進(jìn)入AnonymousObservableSink中調(diào)用run方法,return parent._subscribeHandler(AnyObserver(self))
-->調(diào)用了中保存的閉包回調(diào)_subscribeHandler，傳入了AnyObserver(self),也就是傳入了??這個(gè)行代碼創(chuàng)建的observer
-->直到這個(gè)時(shí)候， let ob = Observable<Any>.create { (obserber) -> Disposable in}中的obserber就被傳入進(jìn)來(lái)
-->當(dāng)obserber調(diào)用了，onNext()、onError()、onCompleted()，則通過(guò)_eventHandler閉包回調(diào)到中的onNext、onError、onCompleted
3.銷毀
下面引用下別人的總結(jié)流程圖
細(xì)細(xì)品一品
再結(jié)合這張圖品一品
