RxSwift

RxSwift简介

RxSwift 是 ReactiveX 家族的重要一员, ReactiveX 是 Reactive Extensions 的缩写，一般简写为Rx。ReactiveX 官方给Rx的定义是，Rx是一个使用可观察数据流进行异步编程的编程接口。

ReactiveX 不仅仅是一个编程接口，它是一种编程思想的突破，它影响了许多其它的程序库和框架以及编程语言。它拓展了观察者模式，使你能够自由组合多个异步事件，而不需要去关心线程，同步，线程安全，并发数据以及I/O阻塞。

RxSwift 是 Rx 为 Swift 语言开发的一门函数响应式编程语言，它可以代替iOS系统的 Target Action / 代理 / 闭包 / 通知 / KVO 等等，同时还提供网络、数据绑定、UI事件处理、UI的展示和更新、多线程等等。

函数响应式编程

*和指令式编程相比，函数式编程强调函数的计算比指令的执行重要。
*和过程化编程相比，函数式编程里函数的计算可随时调用。

RxSwift优势

swift为值类型，在传值与方法回调上有影响，RxSwift一定程度上弥补swift的灵活性

RxSwift使得代码复用性较强，减少代码量

RxSwift因为声明都是不可变更，增加代码可读性

RxSwift使得更易于理解业务代码，抽象异步编程，统一代码风格

RxSwift使得代码更易于编写集成单元测试，增加代码稳定性

RxSwift简单使用

RxSwift-网络请求

    //MARK: - RxSwift应用-网络请求
    func setupNextwork() {
        let url = URL(string: "https://www.baidu.com")
        URLSession.shared.rx.response(request: URLRequest(url: url!))
            .subscribe(onNext: { (response,data) in
                print(response)
        }).disposed(by: disposeBag)
    }

RxSwift-timer定时器

   //MARK: - RxSwift应用-timer定时器
   func setupTimer() {
       timer = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
       timer.subscribe(onNext: { (num) in
           print(num)
       })
       .disposed(by: disposeBag)
   }

RxSwift-通知

    //MARK: - 通知
    func setupNotification(){
        NotificationCenter.default.rx.notification(UIResponder.keyboardWillShowNotification)
            .subscribe(onNext: { (noti) in
                print(noti)
            })
        .disposed(by: disposeBag)
    }

RxSwift-手势

    //MARK: - 手势
    func setupGestureRecognizer(){
        let tap = UITapGestureRecognizer()
        self.label.addGestureRecognizer(tap)
        self.label.isUserInteractionEnabled = true
        tap.rx.event.subscribe(onNext: { (tap) in
            print(tap.view)
        })
        .disposed(by: disposeBag)
    }

RxSwift-UIScrollView

    //MARK: - RxSwift应用-scrollView
    func setupScrollerView() {
        scrollView.rx.contentOffset
            .subscribe(onNext: { [weak self](content) in
                self?.view.backgroundColor = UIColor.init(red: content.y/255*0.8, green: content.y/255*0.6, blue: content.y/255*0.3, alpha: 1)
            })
        .disposed(by: disposeBag)
    }

RxSwift-UITextFiled

    //MARK: - RxSwift应用-textfiled
    func setupTextFiled() {
        self.textFiled.rx.text.orEmpty
            .subscribe(onNext: { (text) in
               print(text)
            })
            .disposed(by: disposeBag)
    }

RxSwift-UIButton

    //MARK: - RxSwift应用-button响应
    func setupButton() {
        self.button.rx.tap
            .subscribe(onNext: { () in
                print("点击来了")
            })
            .disposed(by: disposeBag)
    }

RxSwift-KVO

    //MARK: - RxSwift应用-KVO
    func setupKVO() {
        self.person.rx.observeWeakly(String.self, "name")
            .subscribe(onNext: { (value) in
                print(value as Any)
            })
            .disposed(by: disposeBag)
    }

RxSwift核心逻辑

RxSwift核心流程大致分为：

RxSwift核心逻辑.jpg

1.创建序列：create
2.订阅序列：subscribe
3.发送信号：onNext

let ob = Observable<Any>.create { (obserber) -> Disposable in
            // 3:发送信号
            obserber.onNext("御雪飞斐")
            obserber.onCompleted()
            //obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
            return Disposables.create()
        }

let _ = ob.subscribe(onNext: { (text) in
            print("订阅到:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

核心逻辑分解：

核心逻辑.jpg

创建序列

extension ObservableType {
    public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }
}

final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
    ...
    ...
    func run(_ parent: Parent) -> Disposable {
        return parent._subscribeHandler(AnyObserver(self))
    }
}

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)
        return (sink: sink, subscription: subscription)
    }
}

继承关系如下 AnonymousObservable -> Producer -> Observable -> ObservableType
ObservableType协议中定义的subscribe方法由Producer实现，具体实现有Producer的子类实现
重点代码 let sinkAndSubscription = self.run(observer, cancel: disposer)

class Producer<Element> : Observable<Element> {
   ...
    override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
       ...
        // The returned disposable needs to release all references once it was disposed.
            let disposer = SinkDisposer()
            let sinkAndSubscription = self.run(observer, cancel: disposer)
       ...
    }
}

Observable

public class Observable<Element> : ObservableType {
.....
   public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
        rxAbstractMethod()
    }
.....
}

下面这个方法会在订阅中调用：

 public func asObservable() -> Observable<E> {
        return self
  }

订阅序列subscribe

调用subscribe创建 AnonymousObserver；
AnonymousObserver保存外部的eventHandler；
subscribe中的Disposables调用self.asObservable().subscribe(observer)时，走上面的AnonymousObservable调用父类Producer的subscribe实现流程

extension ObservableType {

public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
           .....
            let observer = AnonymousObserver<E> { event in
              ......
                switch event {
                case .next(let value):
                    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(
                self.asObservable().subscribe(observer),
                disposable
            )
    }
}

AnonymousObserver

final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
    typealias Element = ElementType
    typealias EventHandler = (Event<Element>) -> Void
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
    ....
        self._eventHandler = eventHandler
    }

    override func onCore(_ event: Event<Element>) {
        return self._eventHandler(event)
    }
    .....
}

onNext信号事件

在Controller的代码中：

       let ob = Observable<Any>.create { (obserber) -> Disposable in
            // 3:发送信号
            obserber.onNext("发送信号")
            return Disposables.create()
        }

最终调用到：

extension ObserverType {
    public func onNext(_ element: E) {
        self.on(.next(element))
    }
.......
}

具体实现：
关键代码 ** public func on(_ event: Event<Element>)**

public struct AnyObserver<Element> : ObserverType {
    ......
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }
.....
}

整个RxSwift流程如下：

RxSwift整体流程图.png

RxSwift 笔记一：初探函数响应式编程