RxSwift - 学习笔记一 | 刘峰的个人博客

一、创建可观察者

//
//  ObservableTestVC.swift
//  BasicSwiftDemo
//
//  Created by Pro on 2020/11/11.
//  Copyright © 2020 com.liufeng.mysterFeng. All rights reserved.
//

import UIKit
enum MyError: Error {
    case A
    case B
}
let disposeBag = DisposeBag()


class ObservableTestVC: UIViewController {

    override func viewDidLoad() {
        super.viewDidLoad()

        // 我们可以通过如下几种方法来创建一个 Observable 序列
        
        //传入默认值初始化
        let observable1 = Observable<Int>.just(5)
        observable1.subscribe {
            print($0)
        }.disposed(by: disposeBag)
        
        //必须同类型
        let observable2 = Observable.of("A", "B", "C")
        observable2.subscribe {
            print($0)
        }.disposed(by: disposeBag)
        
        //传入数组
        let observable3 = Observable.from(["A", "B", "C"])
        observable3.subscribe{
         print($0)
        }.disposed(by: disposeBag)
        //创建空内容序列
        let observable4 = Observable<Int>.empty()
        observable4.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        //永远不终止也不发出event
        let observable5 = Observable<Int>.never()
        observable5.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        //不做任何操作直接发出一个错误
        let observable6 = Observable<Int>.error(MyError.A)
        observable6.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        //创建一个范围w数值序列
        let observable7 = Observable.range(start: 1, count: 5)
        observable7.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        //无限发出event永不终止
        let observable8 = Observable.repeatElement(1)
        observable8.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        //满足所有条件发出 初始值是0 必须小于等于10 每次加2
        let observable9 = Observable.generate(
            initialState: 0,
            condition: { $0 <= 10 },
            iterate: { $0 + 2 }
        )
        observable9.subscribe{
         print($0)
        }.disposed(by: disposeBag)

        
        //这个block有一个回调参数observer就是订阅这个Observable对象的订阅者
        let observable10 = Observable<String>.create{observer in
            observer.onNext("啊啊啊啊啊啊啊")
            observer.onCompleted()
            return Disposables.create()
        }
        
        //订阅测试
        observable10.subscribe {
            print($0)
        }.disposed(by: disposeBag)
        
        //用于标记是奇数、还是偶数
        var isOdd = true
        
        //使用deferred()方法延迟Observable序列的初始化，通过传入的block来实现Observable序列的初始化并且返回。
        let factory : Observable<Int> = Observable.deferred {
            
            //让每次执行这个block时候都会让奇、偶数进行交替
            isOdd = !isOdd
            
            //根据isOdd参数，决定创建并返回的是奇数Observable、还是偶数Observable
            if isOdd {
            return Observable.of(1, 3, 5 ,7)
            }else {
                return Observable.of(2, 4, 6, 8)
            }
        }
        
        //第1次订阅测试
        factory.subscribe { event in
            print("\(isOdd)", event)
        }.disposed(by: disposeBag)
        
        //第2次订阅测试
        factory.subscribe { event in
            print("\(isOdd)", event)
        }.disposed(by: disposeBag)
        
        

        //定时器
        
        let observable12 = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        observable12.subscribe { event in
            print(event)
        }.disposed(by: disposeBag)
        
        
        //5秒种后发出唯一的一个元素0
        let observable13 = Observable<Int>.timer(5, scheduler: MainScheduler.instance)
        observable13.subscribe { event in
            print(event)
        }.disposed(by: disposeBag)
        
        
        //延时5秒种后，每隔1秒钟发出一个元素
        let observable14 = Observable<Int>.timer(5, period: 1, scheduler: MainScheduler.instance)
        observable14.subscribe { event in
            print(event)
        }.disposed(by: disposeBag)
        
        _ = Observable<Int>.timer(RxTimeInterval.milliseconds(500), scheduler: MainScheduler.instance)

        
    }
    

}

二、订阅观察者

RxSwift 还提供了另一个 subscribe 方法，它可以把 event 进行分类

通过不同的 block 回调处理不同类型的 event
同时会把 event 携带的数据直接解包出来作为参数，方便我们使用。

所以我们也可以只处理 onNext 而不管其他的情况

let observable = Observable.of("A", "B", "C")
         
observable.subscribe(onNext: { element in
    print(element)
}, onError: { error in
    print(error)
}, onCompleted: {
    print("completed")
}, onDisposed: {
    print("disposed")
})

三、监听事件doOn

let observable = Observable.of("A", "B", "C")
 
observable
    .do(onNext: { element in
        print("Intercepted Next：", element)
    }, onError: { error in
        print("Intercepted Error：", error)
    }, onCompleted: {
        print("Intercepted Completed")
    }, onDispose: {
        print("Intercepted Disposed")
    })
    .subscribe(onNext: { element in
        print(element)
    }, onError: { error in
        print(error)
    }, onCompleted: {
        print("completed")
    }, onDisposed: {
        print("disposed")
    })

四. 观察者（Observer）

观察者（Observer）的作用就是监听事件，然后对这个事件做出响应。或者说任何响应事件的行为都是观察者。

观察者就是由 onNext，onError，onCompleted 这些闭包构建

在 bind 方法中创建

        //Observable序列（每隔1秒钟发出一个索引数）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
 
        observable
            .map { "当前索引数：\($0 )"}
            .bind { [weak self](text) in
                //收到发出的索引数后显示到label上
                self?.label.text = text
            }
            .disposed(by: disposeBag)

AnyObserver 创建观察者

//观察者
let observer: AnyObserver<String> = AnyObserver { (event) in
    switch event {
    case .next(let data):
        print(data)
    case .error(let error):
        print(error)
    case .completed:
        print("completed")
    }
}
 
let observable = Observable.of("A", "B", "C")
observable.subscribe(observer)

配合 bindTo 方法使用

        
        //观察者
        let observer: AnyObserver<String> = AnyObserver { [weak self] (event) in
            switch event {
            case .next(let text):
                //收到发出的索引数后显示到label上
                self?.label.text = text
            default:
                break
            }
        }
         
        //Observable序列（每隔1秒钟发出一个索引数）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        observable
            .map { "当前索引数：\($0 )"}
            .bind(to: observer)
            .disposed(by: disposeBag)

使用 Binder 创建观察者
Binder 主要有以下两个特征：

1.不会处理错误事件

2.确保绑定都是在给定 Scheduler 上执行（默认 MainScheduler）

//观察者
        let observer: Binder<String> = Binder(label) { (view, text) in
            //收到发出的索引数后显示到label上
            view.text = text
        }
         
        //Observable序列（每隔1秒钟发出一个索引数）
        let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        observable
            .map { "当前索引数：\($0 )"}
            .bind(to: observer)
            .disposed(by: disposeBag)
    }

Binder 在 RxCocoa 中的应用

extension Reactive where Base: UIControl {
     
    /// Bindable sink for `enabled` property.
    public var isEnabled: Binder<Bool> {
        return Binder(self.base) { control, value in
            control.isEnabled = value
        }
    }
}

let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
observable
    .map { $0 % 2 == 0 }
    .bind(to: button.rx.isEnabled)
    .disposed(by: disposeBag)

五、自定义可绑定属性

1.对UI控件自定义

extension UILabel {
    public var fontSize: Binder<CGFloat> {
        return Binder(self) { label, fontSize in
            label.font = UIFont.systemFont(ofSize: fontSize)
        }
    }
}

2.通过对 Reactive 类进行扩展

extension Reactive where Base: UILabel {
    public var fontSize: Binder<CGFloat> {
        return Binder(self.base) { label, fontSize in
            label.font = UIFont.systemFont(ofSize: fontSize)
        }
    }
}

RxSwift 已经为我们提供许多常用的可绑定属性,自己阅读即可

六、Subjects 介绍

当我们创建一个 Observable 的时候就要预先将要发出的数据都准备好，等到有人订阅它时再将数据通过 Event 发出去。
但有时我们希望 Observable 在运行时能动态地“获得”或者说“产生”出一个新的数据，再通过 Event 发送出去。比如：订阅一个输入框的输入内容，当用户每输入一个字后，这个输入框关联的 Observable 就会发出一个带有输入内容的 Event，通知给所有订阅者

Subjects 既是订阅者，也是 Observable

说它是订阅者，是因为它能够动态地接收新的值
说它又是一个 Observable，是因为当 Subjects 有了新的值之后，就会通过 Event 将新值发出给他的所有订阅者

Subject 常用的几个方法

onNext(:)：是 on(.next(:)) 的简便写法。该方法相当于 subject 接收到一个 .next 事件。
onError(:)：是 on(.error(:)) 的简便写法。该方法相当于 subject 接收到一个 .error 事件。
onCompleted()：是 on(.completed) 的简便写法。该方法相当于 subject 接收到一个 .completed 事件

一共有四种 Subjects

PublishSubject
BehaviorSubject
ReplaySubject
Variable

七、变换操作
buffer

buffer 方法作用是缓冲组合，第一个参数是缓冲时间，第二个参数是缓冲个数，第三个参数是线程。
window

window 操作符和 buffer 十分相似。不过 buffer 是周期性的将缓存的元素集合发送出来，而 window 周期性的将元素集合以 Observable 的形态发送出来。
map

该操作符通过传入一个函数闭包把原来的 Observable 序列转变为一个新的 Observable 序列。
flatMap

map 在做转换的时候容易出现“升维”的情况。即转变之后，从一个序列变成了一个序列的序列。
而 flatMap 操作符会对源 Observable 的每一个元素应用一个转换方法，将他们转换成 Observables。然后将这些 Observables 的元素合并之后再发送出来。即又将其 “拍扁”（降维）成一个 Observable 序列。
这个操作符是非常有用的。比如当 Observable 的元素本生拥有其他的 Observable 时，我们可以将所有子 Observables 的元素发送出来
flatMapLatest

flatMapLatest 与 flatMap 的唯一区别是：flatMapLatest 只会接收最新的 value 事件。
flatMapFirst

flatMapFirst 与 flatMapLatest 正好相反：flatMapFirst 只会接收最初的 value 事件。
concatMap

concatMap 与 flatMap 的唯一区别是：当前一个 Observable 元素发送完毕后，后一个Observable 才可以开始发出元素。或者说等待前一个 Observable 产生完成事件后，才对后一个 Observable 进行订阅。
scan

scan 就是先给一个初始化的数，然后不断的拿前一个结果和最新的值进行处理操作。
groupBy

groupBy 操作符将源 Observable 分解为多个子 Observable，然后将这些子 Observable 发送出来。
也就是说该操作符会将元素通过某个键进行分组，然后将分组后的元素序列以 Observable 的形态发送出来。
filter

该操作符就是用来过滤掉某些不符合要求的事件。
take

该方法实现仅发送 Observable 序列中的前 n 个事件，在满足数量之后会自动 .completed。
skip

该方法用于跳过源 Observable 序列发出的前 n 个事件。
distinctUntilChanged

该操作符用于过滤掉连续重复的事件。
single

限制只发送一次事件，或者满足条件的第一个事件。
如果存在有多个事件或者没有事件都会发出一个 error 事件。
如果只有一个事件，则不会发出 error 事件。
elementAt

该方法实现只处理在指定位置的事件。
ignoreElements

该操作符可以忽略掉所有的元素，只发出 error 或 completed 事件。
如果我们并不关心 Observable 的任何元素，只想知道 Observable 在什么时候终止，那就可以使用 ignoreElements 操作符。
takeLast

该方法实现仅发送 Observable 序列中的后 n 个事件。
Sample

Sample 除了订阅源 Observable 外，还可以监视另外一个 Observable，即 notifier 。
每当收到 notifier 事件，就会从源序列取一个最新的事件并发送。而如果两次 notifier 事件之间没有源序列的事件，则不发送值。
debounce

debounce 操作符可以用来过滤掉高频产生的元素，它只会发出这种元素：该元素产生后，一段时间内没有新元素产生。
换句话说就是，队列中的元素如果和下一个元素的间隔小于了指定的时间间隔，那么这个元素将被过滤掉。
debounce 常用在用户输入的时候，不需要每个字母敲进去都发送一个事件，而是稍等一下取最后一个事件。
amb

当传入多个 Observables 到 amb 操作符时，它将取第一个发出元素或产生事件的 Observable，然后只发出它的元素。并忽略掉其他的 Observables。
takeWhile

该方法依次判断 Observable 序列的每一个值是否满足给定的条件。当第一个不满足条件的值出现时，它便自动完成。
takeUntil

除了订阅源 Observable 外，通过 takeUntil 方法我们还可以监视另外一个 Observable，即 notifier。
如果 notifier 发出值或 complete 通知，那么源 Observable 便自动完成，停止发送事件。
skipWhile

该方法用于跳过前面所有满足条件的事件。
一旦遇到不满足条件的事件，之后就不会再跳过了
skipUntil

同上面的 takeUntil 一样，skipUntil 除了订阅源 Observable 外，通过 skipUntil 方法我们还可以监视另外一个 Observable，即 notifier 。
与 takeUntil 相反的是。源 Observable 序列事件默认会一直跳过，直到 notifier 发出值或 complete 通知。
startWith

该方法会在 Observable 序列开始之前插入一些事件元素。即发出事件消息之前，会先发出这些预先插入的事件消息。
merge

该方法可以将多个（两个或两个以上的）Observable 序列合并成一个 Observable 序列
zip

该方法可以将多个（两个或两个以上的）Observable 序列压缩成一个 Observable 序列。
而且它会等到每个 Observable 事件一一对应地凑齐之后再合并。
比如我们想同时发送两个请求，只有当两个请求都成功后，再将两者的结果整合起来继续往下处理。这个功能就可以通过 zip 来实现。

//第一个请求
let userRequest: Observable<User> = API.getUser("me")
 
//第二个请求
let friendsRequest: Observable<Friends> = API.getFriends("me")
 
//将两个请求合并处理
Observable.zip(userRequest, friendsRequest) {
        user, friends in
        //将两个信号合并成一个信号，并压缩成一个元组返回（两个信号均成功）
        return (user, friends)
    }
    .observeOn(MainScheduler.instance) //加这个是应为请求在后台线程，下面的绑定在前台线程。
    .subscribe(onNext: { (user, friends) in
        //将数据绑定到界面上
        //.......
    })
    .disposed(by: disposeBag)

combineLatest

该方法同样是将多个（两个或两个以上的）Observable 序列元素进行合并。
但与 zip 不同的是，每当任意一个 Observable 有新的事件发出时，它会将每个 Observable 序列的最新的一个事件元素进行合并
withLatestFrom

该方法将两个 Observable 序列合并为一个。每当 self 队列发射一个元素时，便从第二个序列中取出最新的一个值
switchLatest

switchLatest 有点像其他语言的 switch 方法，可以对事件流进行转换。
比如本来监听的 subject1，我可以通过更改 variable 里面的 value 更换事件源。变成监听 subject2。
toArray

该操作符先把一个序列转成一个数组，并作为一个单一的事件发送，然后结束。
reduce

reduce 接受一个初始值，和一个操作符号。
reduce 将给定的初始值，与序列里的每个值进行累计运算。得到一个最终结果，并将其作为单个值发送出去。
concat

concat 会把多个 Observable 序列合并（串联）为一个 Observable 序列。
并且只有当前面一个 Observable 序列发出了 completed 事件，才会开始发送下一个 Observable 序列事件

RxSwift - 学习笔记一