Operator

Operator - 操作符

操作符可以帮助创建新的序列，或者变化组合原有的序列，从而生成一个新的序列。

例如，通过 map 方法将输入的用户名，转换为用户名是否有效。然后用这个转化后来的序列来控制红色提示语是否隐藏。我们还通过 combineLatest 方法，将用户名是否有效和密码是否有效合并成两者是否同时有效。然后用这个合成后来的序列来控制按钮是否可点击。

这里 map 和 combineLatest 都是操作符，它们可以帮助我们构建所需要的序列

filter - 过滤

可以用 filter 创建一个新的序列。这个序列只发出温度大于 33 度的元素。

map - 转换

可以用 map 创建一个新的序列。这个序列将原有的 JSON 转换成 Model 。这种转换实际上就是解析 JSON 。

zip - 配对

可以用 zip 来合成一个新的序列。这个序列将汉堡序列的元素和薯条序列的元素配对后，生成一个新的套餐序列。

• 如何使用操作符

使用操作符是非常容易的。可以直接调用实例方法，或者静态方法

// 温度
let rxTemperature: Observable<Double> = ...

// filter 操作符
rxTemperature.filter { temperature in temperature > 33 }
    .subscribe(onNext: { temperature in
        print("高温：(temperature)度")
    })
    .disposed(by: disposeBag)

// JSON
let json: Observable<JSON> = ...

// map 操作符
json.map(Model.init)
    .subscribe(onNext: { model in
        print("取得 Model: (model)")
    })
    .disposed(by: disposeBag)

// 汉堡
let rxHamburg: Observable<Hamburg> = ...
// 薯条
let rxFrenchFries: Observable<FrenchFries> = ...

// zip 操作符
Observable.zip(rxHamburg, rxFrenchFries)
    .subscribe(onNext: { (hamburg, frenchFries) in
        print("取得汉堡: (hamburg) 和薯条：(frenchFries)")
    })
    .disposed(by: disposeBag)

Rx 提供了充分的操作符来帮我们创建序列。当然如果内置操作符无法满足你的需求时，还可以创建自定义的操作符。

操作符列表

• amb
• buffer
• catchError
• combineLatest
• concat
• concatMap
• connect
• create
• debounce
• debug
• deferred
• delay
• delaySubscription
• dematerialize
• distinctUntilChanged
• do
• elementAt
• empty
• error
• filter
• flatMap
• flatMapLatest
• from
• groupBy
• ignoreElements
• interval
• just
• map
• merge
• materialize
• never
• observeOn
• publish
• reduce
• refCount
• repeatElement
• replay
• retry
• sample
• scan
• shareReplay
• single
• skip
• skipUntil
• skipWhile
• startWith
• subscribeOn
• take
• takeLast
• takeUntil
• takeWhile
• timeout
• timer
• using
• window
• withLatestFrom
• zip

Disposable - 可被清除的资源

通常来说，一个序列如果发出了 error 或者 completed 事件，那么所有内部资源都会被释放。如果需要提前释放这些资源或取消订阅的话，那么可以对返回的 可被清除的资源（Disposable） 调用 dispose 方法：

ar disposable: Disposable?

override func viewWillAppear(_ animated: Bool) {
    super.viewWillAppear(animated)

    self.disposable = textField.rx.text.orEmpty
        .subscribe(onNext: { text in print(text) })
}

override func viewWillDisappear(_ animated: Bool) {
    super.viewWillDisappear(animated)

    self.disposable?.dispose()
}

调用 dispose 方法后，订阅将被取消，并且内部资源都会被释放。通常情况下，是不需要手动调用 dispose 方法的。推荐使用 清除包（DisposeBag） 或者 takeUntil 操作符 来管理订阅的生命周期。

• DisposeBag - 清除包

使用 ARC 来管理内存。那么我们能不能用 ARC 来管理订阅的生命周期了。答案是肯定了，可以用 清除包（DisposeBag） 来实现这种订阅管理机制：

var disposeBag = DisposeBag()

override func viewWillAppear(_ animated: Bool) {
    super.viewWillAppear(animated)

    textField.rx.text.orEmpty
        .subscribe(onNext: { text in print(text) })
        .disposed(by: self.disposeBag)
}

override func viewWillDisappear(_ animated: Bool) {
    super.viewWillDisappear(animated)

    self.disposeBag = DisposeBag()
}

当 清除包 被释放的时候，清除包 内部所有 可被清除的资源（Disposable） 都将被清除

var disposeBag = DisposeBag() // 来自父类 ViewController

override func viewDidLoad() {
    super.viewDidLoad()

    ...

    usernameValid
        .bind(to: passwordOutlet.rx.isEnabled)
        .disposed(by: disposeBag)

    usernameValid
        .bind(to: usernameValidOutlet.rx.isHidden)
        .disposed(by: disposeBag)

    passwordValid
        .bind(to: passwordValidOutlet.rx.isHidden)
        .disposed(by: disposeBag)

    everythingValid
        .bind(to: doSomethingOutlet.rx.isEnabled)
        .disposed(by: disposeBag)

    doSomethingOutlet.rx.tap
        .subscribe(onNext: { [weak self] in self?.showAlert() })
        .disposed(by: disposeBag)
}

disposeBag 和 ViewController 具有相同的生命周期。当退出页面时， ViewController 就被释放，disposeBag 也跟着被释放了，那么这里的 5 次绑定（订阅）也就被取消了。这正是我们所需要的。

takeUntil

另外一种实现自动取消订阅的方法就是使用 takeUntil 操作符，上面那个输入验证的演示代码也可以通过使用 takeUntil 来实现：

override func viewDidLoad() {
    super.viewDidLoad()

    ...

    _ = usernameValid
        .takeUntil(self.rx.deallocated)
        .bind(to: passwordOutlet.rx.isEnabled)

    _ = usernameValid
        .takeUntil(self.rx.deallocated)
        .bind(to: usernameValidOutlet.rx.isHidden)

    _ = passwordValid
        .takeUntil(self.rx.deallocated)
        .bind(to: passwordValidOutlet.rx.isHidden)

    _ = everythingValid
        .takeUntil(self.rx.deallocated)
        .bind(to: doSomethingOutlet.rx.isEnabled)

    _ = doSomethingOutlet.rx.tap
        .takeUntil(self.rx.deallocated)
        .subscribe(onNext: { [weak self] in self?.showAlert() })
}

这将使得订阅一直持续到控制器的 dealloc 事件产生为止。

Schedulers - 调度器

Schedulers 是 Rx 实现多线程的核心模块，它主要用于控制任务在哪个线程或队列运行。

如果你曾经使用过 GCD， 那你对以下代码应该不会陌生：

// 后台取得数据，主线程处理结果
DispatchQueue.global(qos: .userInitiated).async {
    let data = try? Data(contentsOf: url)
    DispatchQueue.main.async {
        self.data = data
    }
}

如果用 RxSwift 来实现，大致是这样的：

let rxData: Observable<Data> = ...

rxData
    .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated))
    .observeOn(MainScheduler.instance)
    .subscribe(onNext: { [weak self] data in
        self?.data = data
    })
    .disposed(by: disposeBag)

使用 subscribeOn

我们用 subscribeOn 来决定数据序列的构建函数在哪个 Scheduler 上运行。以上例子中，由于获取 Data 需要花很长的时间，所以用 subscribeOn 切换到 后台 Scheduler 来获取 Data。这样可以避免主线程被阻塞。

使用 observeOn

我们用 observeOn 来决定在哪个 Scheduler 监听这个数据序列。以上例子中，通过使用 observeOn 方法切换到主线程来监听并且处理结果。

一个比较典型的例子就是，在后台发起网络请求，然后解析数据，最后在主线程刷新页面。你就可以先用 subscribeOn 切到后台去发送请求并解析数据，最后用 observeOn 切换到主线程更新页面。

MainScheduler

MainScheduler 代表主线程。如果你需要执行一些和 UI 相关的任务，就需要切换到该 Scheduler 运行。

SerialDispatchQueueScheduler

SerialDispatchQueueScheduler 抽象了串行 DispatchQueue。如果你需要执行一些串行任务，可以切换到这个 Scheduler 运行。

ConcurrentDispatchQueueScheduler

ConcurrentDispatchQueueScheduler 抽象了并行 DispatchQueue。如果你需要执行一些并发任务，可以切换到这个 Scheduler 运行。

OperationQueueScheduler

OperationQueueScheduler 抽象了 NSOperationQueue。

它具备 NSOperationQueue 的一些特点，例如，你可以通过设置 maxConcurrentOperationCount，来控制同时执行并发任务的最大数量。