目的

gitHub地址: https://github.com/ReactiveX/RxSwift

RxSwift的目的是让数据/事件流和异步任务能够更方便的序列化处理， 能够使用Swift进行响应式编程。

RxSwift做了什么

RxSwift把我们程序中每一个操作都看成一个事件，比如一个TextField中的文本改变，一个按钮被点击，或者一个网络请求结束等，每一个事件源就可以看成一个管道，也就是sequence，比如TextField，当我们改变里面的文本的时候，这个TextField就会不断的发出事件，从他的这个sequence中不断的流出，我们只需要监听这个sequence，每流出一个事件就做相应的处理。同理，Button也是一个sequence，每点击一次就流出一个事件。

RxSwift的核心思想是 Observable

sequence，observable表示可监听或者可观察，也就是说RxSwift的核心思想是可监听的序列。并且Observable sequence可以接受异步信号，也就是说，信号是可以异步给监听者的，Observable(ObservableType) 和 SequenceType类似，ObservableType.subscribe 和 SequenceType.generate类似，由于RxSwift支持异步获得信号，所以用ObservableType.subscribe，这和indexGenerator.next()类似其中SequenceType是Swift(2.3以前版本，之后的版本没有该协议)中的一个协议，比如Swift中的Array就遵循这个协议，通过这个协议，你可以这样的去操作一个Array

let array = [1, 2, 3, 4, 5]

let array2 = array.filter({$0 > 1}).map({$0 * 2})//4 6 8 10

var indexGenerator = array2.generate()

let fisrt = indexGenerator.next() // 4

let seoncd = indexGenerator.next() //6

RxSwift中，ObservableType.subscribe的回调（新的信号到来）一共有三个

enum Event<Element> {
    
    case Next(Element) // 新的信号到来
    
    case Error(ErrorType) // 信号发生错误，序列不会再产生信号
    
    case Completed // 序列发送信号完成，不会再产生新的信号
    
}

Observable分为两种

1 在有限的时间内会自动结束（Completed/Error），比如一个网络请求当作一个序列，当网络请求完成的时候，Observable自动结束，资源会被释放。
2 信号不会自己结束，最简单的比如一个Timer，每隔一段时间发送一个新的信号过来，这时候需要手动取消监听，来释放相应的资源，比如一个label.rx.text是一个Obserable，通常需要这样调用addDisposableTo(disposeBag)来让其在deinit，也就是所有者要释放的时候，自动取消监听。

class Observable<Element> {
    func subscribe(observer: Observer<Element>) -> Disposable //调用Disposable的方法来取消
}

当然，除了手动释放，RxSwift提供了一些操作符，比如 takeUntil来根据条件取消

sequence .takeUntil(self.rx__deallocated) //当对象要释放的时候，取消监听 .subscribe { print($0) }

Operator运算符相关

never

never就是创建一个sequence，但是不发出任何事件信号。

Observable<String>.never().subscribe { _ in
     print("This will never be printed")
}.disposed(by: disposeBag)

empty

empty就是创建一个空的sequence，只能发出一个complected事件。

Observable<Int>.empty().subscribe { event in
     let text = "operator: empty__\(event)"
     self.showText(text: text)
     print(text)
}.disposed(by: disposeBag)

create

我们也可以自定义可观察的sequence，那就是使用create。
create操作符传入一个观察者observer，然后调用observer的onNext，onCompleted和onError方法。返回一个可观察的obserable序列。
无参创建create

Observable<Any>.create { (observal:AnyObserver<Any>) -> Disposable in
    observal.onNext("abc")
    observal.onNext("12")
    observal.onCompleted()
    return Disposables.create()
}.subscribe(onNext: { str in
    let text = "operator: create__\(str)"
    self.showText(text: text)
    print(text)
}).disposed(by: disposeBag)

添加参数创建create

func createObservable(element:String) -> Observable<String> {
    return Observable.create { (observal:AnyObserver<String>) -> Disposable in
                observal.onNext(element)
                observal.onCompleted()
           return Disposables.create()
   }
}

createObservable(element: "element").subscribe(onNext: { str in
       let text = "operator: create with element__\(str)"
       self.showText(text: text)
       print(text)
}).disposed(by: disposeBag)

deferred

延时创建Observable对象，当subscribe的时候才去创建，它为每一个Observer创建一个新的Observable，也就是说每个订阅者订阅的对象都是内容相同但是完全独立的序列。
deferr采用一个Factory函数型作为参数，Factory函数返回的是Observable类型。这也是其延时创建Observable的主要实现。

let defObservable = Observable<String>.deferred { () -> Observable<String> in
       return Observable.create { observer -> Disposable in
                   observer.onNext("defObservable create")
                   observer.onCompleted()
              return Disposables.create()
       }
}
defObservable.subscribe { event in
       let text = "operator: deferred one__\(event)"
       self.showText(text: text)
       print(text)
}.disposed(by: disposeBag)
defObservable.subscribe { event in
       let text = "operator: deferred two__\(event)"
       self.showText(text: text)
       print(text)
}.disposed(by: disposeBag)

/// deferred 延迟创建的例子
var value1:String?
let observable1 = Observable<String>.from(optional: value1)
value1 = "Darren1"
observable1.subscribe { event in
                       
     /// 只打印出 completed
     /// 并没有像我们想象中的那样也会打印出 onNext事件，这个是为什么呢？因为在我们订阅的时候，数据未必已经初始化完成
     let text = "operator: deferred__\(event)"
     self.showText(text: text)
     print(text)
}.disposed(by: disposeBag)

/// 把这个例子使用defer重新测试一下
var value2:String?
let observable2 = Observable<String>.deferred { () -> Observable<String> in
      return Observable<String>.from(optional: value2)
}
value2 = "Darren2"
observable2.subscribe { event in
     let text = "operator: deferred__\(event)"
     self.showText(text: text)
                       
     /// next(Darren2) completed
     print(text)
}.disposed(by: disposeBag)

of/from

这两个方法都是把一个对象或者数据转换为可观察序列，这在使用Swift中的SequenceType时很有用。

Observable<String>.of("hello", "RxSwift").subscribe { event in
     let text = "operator: of__\(event)"
     self.showText(text: text)
                                                     
     /// next(hello) next(RxSwift) completed
     print(text)
}.disposed(by: disposeBag)
Observable<String>.from(["hello", "RxSwift"]).subscribe { event in
     let text = "operator: from__\(event)"
     self.showText(text: text)
                                                         
     /// next(hello) next(RxSwift) completed
     print(text)
}.disposed(by: disposeBag)

just

将一个对象或者一个Sequence转换为 一个可观察序列，请注意这里与From是完全不相同的：from是转换为一个或者多个可观察序列（这取决于你是要将一个还是一个序列进行转换）。也就是说just只能包含一个观察序列，请注意与上面例子结果进行对比。

Observable<Array<String>>.just(["hello", "RxSwift"]).subscribe { event in
     let text = "operator: just__\(event)"
     self.showText(text: text)
                                                                
     /// next(["hello", "RxSwift"]) completed
     print(text)
}.disposed(by: disposeBag)

range

给定范围，依次显示，就是创建一个sequence，他会发出这个范围中的从开始到结束的所有事件，Observable必须指定数据类型。

Observable<Int>.range(start: 1, count: 4).subscribe(onNext: { value in
     let text = "operator: range__\(value)"
     self.showText(text: text)
     print(text)
}).disposed(by: disposeBag)

interval

创建一个可观察序列，以特定的时间间隔释放一系列整数(E -> Int/NSInteger)。

Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).take(3).subscribe { event in
     let text = "operator: interval__\(event)"
     self.showText(text: text)
                                                                                                          
     /// operator: interval__next(0)
     /// operator: interval__next(1)
     /// operator: interval__next(2)
     /// operator: interval__completed
     print(text)
}.disposed(by: disposeBag)

timer

在指定的时间后，发送一个特定的item (E -> Int/NSInteger)，请注意这里与interval的区别（interval是发送一系列特定item，而timer只会发送一个）。

Observable<Int>.timer(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe { event in
     let text = "operator: timer__\(event)"
     self.showText(text: text)
                                                                                               
     /// operator: timer__next(0)
     /// operator: timer__completed
     print(text)
}.disposed(by: disposeBag)

repeatElement

创建一个sequence，发出特定的事件n次。

/// 如果没有.take(3), 就会一直执行下去
Observable.repeatElement("darren").take(3).subscribe(onNext: { value in
     let text = "operator: repeat__\(value)"
     self.showText(text: text)
     print(text)
}).disposed(by: disposeBag)

generate

类似于for循环，创建一个可观察sequence，当初始化的条件为true的时候，他就会发出所对应的事件。

Observable.generate(initialState: 0) {
    $0 < 5
} iterate: {
    $0 + 2
}.subscribe(onNext: { value in
     let text = "operator: generate__\(value)"
     self.showText(text: text)
     print(text)
}).disposed(by: disposeBag)

error

发出错误信号，创建一个可观察序列，但不发出任何正常的事件，只发出error事件并结束。

let error = NSError(domain: "error", code: 10, userInfo: ["This is error" : "xxxxxx"]) as Error
Observable<Any>.error(error).subscribe(onNext: { value in
     let text = "operator: error__\(value)"
     self.showText(text: text)
     print(text)
}).disposed(by: disposeBag)

Transform 变换相关

buffer

定期的将需要发射的items随机到一个buffer的包中，分批次的发射这些包，而不是一次发射一个item：例如你有[1, 2, 3, 4] ，你可以一次发射一个，也可以一次发射两个item或者三个。

/// 一次发射1个Item事件
Observable<Int>.of(1, 2, 3, 4).buffer(timeSpan: RxTimeInterval.seconds(1), count: 1, scheduler: MainScheduler.instance).subscribe { event in
     let text = "transform: buffer__\(event)"
     self.showText(text: text)
                                     
     /// transform: buffer__next([1])
     /// transform: buffer__next([2])
     /// transform: buffer__next([3])
     /// transform: buffer__next([4])
     /// transform: buffer__next([])
     /// transform: buffer__completed
     print(text)
}.disposed(by: disposeBag)

/// 一次发射3个Item事件
Observable<Int>.of(1, 2, 3, 4).buffer(timeSpan: RxTimeInterval.seconds(1), count: 3, scheduler: MainScheduler.instance).subscribe { event in
     let text = "transform: buffer__\(event)"
     self.showText(text: text)
     
     /// transform: buffer__next([1, 2, 3])
     /// transform: buffer__next([4])
     /// transform: buffer__completed
     print(text)
}.disposed(by: disposeBag)

window

与buffer类似，但是每次发射的不是item，而是Observables序列（请注意与buffer的结果比较）。

Observable<Int>.of(1, 2, 3, 4).window(timeSpan: RxTimeInterval.seconds(1), count: 3, scheduler: MainScheduler.instance).subscribe { event in
     let text = "transform: window__\(event)"
     self.showText(text: text)
                                                                                                                                   
     /// transform: window__next(RxSwift.AddRef<Swift.Int>)
     /// transform: window__next(RxSwift.AddRef<Swift.Int>)
     /// transform: window__completed
     print(text)
}.disposed(by: disposeBag)       

flatMap

将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable。
这个方法是很有用的，例如当你有一个这样的Observable：它发射一个数据序列，这些数据本身包含Observable成员或者可以变换为Observable，因此你可以创建一个新的Observable发射这些次级Observable发射的数据的完整集合。

/// 我需要在每一个Item后跟一个新的Item叫做RxSwift
Observable<Int>.of(0, 1, 2).flatMap { (element:Int) -> Observable<String> in
     return Observable<String>.of("\(element)", "RxSwift")
}.subscribe { event in
     let text = "transform: flatMap__\(event)"
     self.showText(text: text)
                                                 
     /// transform: flatMap__next(0)
     /// transform: flatMap__next(RxSwift)
     /// transform: flatMap__next(1)
     /// transform: flatMap__next(RxSwift)
     /// transform: flatMap__next(2)
     /// transform: flatMap__next(RxSwift)
     /// transform: flatMap__completed
     print(text)
}.disposed(by: disposeBag)

groupBy

将一个Observable分拆为一些Observables集合，它们中的每一个发射原始Observable的一个子序列。

/// 我需要将奇数偶数分成两组
Observable<Int>.of(0, 1, 2, 3, 4, 5).groupBy { (element) -> String in
     return element % 2 == 0 ? "偶数" : "奇数"
}.subscribe { event in
     switch event {
            case .next(let group):
                 group.asObservable().subscribe { event in
                        let text = "transform: groupBy__key:  \(group.key)  \(event)"
                        self.showText(text: text)
                                                                                              
                        /// transform: groupBy__ key:  偶数  next(0)
                        /// transform: groupBy__ key:  奇数  next(1)
                        /// transform: groupBy__ key:  偶数  next(2)
                        /// transform: groupBy__ key:  奇数  next(3)
                        /// transform: groupBy__ key:  偶数  next(4)
                        /// transform: groupBy__ key:  奇数  next(5)
                        /// transform: groupBy__ key:  偶数  completed
                        /// transform: groupBy__ key:  奇数  completed
                        print(text)
                 }.disposed(by: self.disposeBag)
                                                              
            default:break
     }
}.disposed(by: disposeBag)

map

通过一个闭包将原来的序列转换为一个新序列的操作。

Observable<Int>.of(1, 2, 3).map { return "hello " + "\($0)" }.subscribe { event in
     let text = "transform: map__\(event)"
     self.showText(text: text)
                                                                         
     /// transform: map__next(hello 1)
     /// transform: map__next(hello 2)
     /// transform: map__next(hello 3)
     /// transform: map__completed
     print(text)
}.disposed(by: disposeBag)

scan

从字面意思可以看出是扫描，也就是说该方法会给出一个初始值（seed)，每次通过一个函数将上一次的结果与序列中的Item进行处理，每处理完成都会发射.next事件。

Observable<String>.of("Rx", "Swift").scan("hello ") { acum, element in
     return acum + element
}.subscribe { event in
     let text = "transform: scan__\(event)"
     self.showText(text: text)
                                                                 
     /// transform: scan__next(hello Rx)
     /// transform: scan__next(hello RxSwift)
     /// transform: scan__completed
     print(text)
}.disposed(by: disposeBag)

reduce

与上述scan类似，都是初始一个seed，每次通过函数将上一次的结果与序列中的item进行处理，但是唯一不同的一点是，只会在最后发射一次.next事件，将其拿来作数学计算很有用，这个我们将会在后面讲到 （请注意与上述scan的结果比较）。

Observable<String>.of("Rx", "Swift").reduce("hello ") { acum, element in
     return acum + element
}.subscribe { event in
    let text = "transform: reduce__\(event)"
    self.showText(text: text)
                                                                   
    /// transform: reduce__next(hello RxSwift)
    /// transform: reduce__completed
    print(text)
}.disposed(by: disposeBag)

Filter过滤器相关

debounce

在规定的时间内过滤item。

Observable<Int>.of(1, 2, 3, 4).debounce(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe { event in
     let text = "filter: debounce__\(event)"
     self.showText(text: text)
                                                                                                                 
     /// filter: debounce__next(4)
     /// filter: debounce__completed
     print(text)
}.disposed(by: disposeBag)

distinctUntilChanged

过滤掉可观察到的重复item，表示如果发射的事件与上一次不相同那么才会发射此次事件。

Observable<Int>.of(1, 2, 2, 2, 3).distinctUntilChanged().subscribe { event in
     let text = "filter: distinctUntilChanged__\(event)"
     self.showText(text: text)
                                                                    
     /// filter: distinctUntilChanged__next(1)
     /// filter: distinctUntilChanged__next(2)
     /// filter: distinctUntilChanged__next(3)
     /// filter: distinctUntilChanged__completed
     print(text)
}.disposed(by: disposeBag)

elementAt

发射第 n个item。

Observable<Int>.of(1, 2, 3, 4, 5).element(at: 3).subscribe { event in
     let text = "filter: elementAt__\(event)"
     self.showText(text: text)
                                                            
     /// filter: elementAt__next(4)
     /// filter: elementAt__completed
     print(text)
}.disposed(by: disposeBag)

filter

仅发射谓词测试通过的Items。

Observable<Int>.of(9, 10, 11, 12).filter { element -> Bool in
     return element > 10
}.subscribe { event in
     let text = "filter: filter__\(event)"
     self.showText(text: text)
                                                      
     /// filter: filter__next(11)
     /// filter: filter__next(12)
     /// filter: filter__completed
     print(text)
}.disposed(by: disposeBag)

skip

发射第n（包含n）之后的items。

Observable<Int>.of(9, 10, 11, 12).skip(2).subscribe { event in
     let text = "filter: skip__\(event)"
     self.showText(text: text)
                                                     
     /// filter: skip__next(11)
     /// filter: skip__next(12)
     /// filter: skip__completed
     print(text)
}.disposed(by: disposeBag)

take

发射第n（不包含n）之前的items，与skip相反效果。

Observable<Int>.of(9, 10, 11, 12).take(2).subscribe { event in
     let text = "filter: take__\(event)"
     self.showText(text: text)
                                                     
     /// filter: take__next(9)
     /// filter: take__next(10)
     /// filter: take__completed
     print(text)
}.disposed(by: disposeBag)

takeLast

发射第n（包含n）之后的items，与skip相同效果。

Observable<Int>.of(9, 10, 11, 12).takeLast(2).subscribe { event in
     let text = "filter: takeLast__\(event)"
     self.showText(text: text)
                                                         
     /// filter: takeLast__next(11)
     /// filter: takeLast__next(12)
     /// filter: takeLast__completed
     print(text)
}.disposed(by: disposeBag)

Combine结合相关

merge

将多个序列的items合并为一个序列的items。

let observable1 = Observable<Int>.of(1, 3, 5, 7, 9)
let observable2 = Observable<Int>.of(2, 4, 6)
Observable<Int>.merge(observable1, observable2).subscribe { event in
     let text = "combine: merge__\(event)"
     self.showText(text: text)
                                                           
     /// combine: merge__next(1)
     /// combine: merge__next(2)
     /// combine: merge__next(3)
     /// combine: merge__next(4)
     /// combine: merge__next(5)
     /// combine: merge__next(6)
     /// combine: merge__next(7)
     /// combine: merge__next(9)
     /// combine: merge__completed
     print(text)
}.disposed(by: disposeBag)

startWith

在发射序列items前新增一个item。

Observable<String>.of("  ", "RxSwift", "!").startWith("hello").reduce("") { (accum, element) -> String in
     return accum + element
}.subscribe { event in
     let text = "combine: startWith__\(event)"
     self.showText(text: text)
                                                                                       
     /// combine: startWith__next(hello  RxSwift!)
     /// combine: startWith__completed
     print(text)
}.disposed(by: disposeBag)

zip

将多个序列的items进行一一合并，但是需要注意的是，它会等到item对其后合并，未对齐的会舍弃。

let observable3 = Observable<Int>.of(1, 2, 3, 4, 5, 6, 7)
let observable4 = Observable<String>.of("A", "B", "C", "D")
Observable<String>.zip(observable3, observable4) { (e1:Int, e2:String) -> String in
     return "\(e1)\(e2)"
}.subscribe { event in
     let text = "combine: zip__\(event)"
     self.showText(text: text)
                                                              
     /// combine: zip__next(1A)
     /// combine: zip__next(2B)
     /// combine: zip__next(3C)
     /// combine: zip__next(4D)
     /// combine: zip__completed
     print(text)
}.disposed(by: disposeBag)

combineLatest

如果存在两条事件队列，需要同时监听，那么每当有新的事件发生的时候，combineLatest 会将每个队列的最新的一个元素进行合并。
类似于zip，但是只有当原始的Observable中的每一个都发射了一条数据时zip才发射数据。
combineLatest则在原始的Observable中任意一个发射了数据时发射一条数据。当原始Observables的任何一个发射了一条数据时，combineLatest使用一个函数结合它们最近发射的数据，然后发射这个函数的返回值。

let observable5 = Observable<Int>.of(1, 2, 3, 4, 5)
let observable6 = Observable<String>.of("A", "B", "C", "D")
Observable<String>.combineLatest(observable5, observable6) { (e1:Int, e2:String) -> String in
     return "\(e1)\(e2)"
}.subscribe { event in
     let text = "combine: combineLatest__\(event)"
     self.showText(text: text)
                                                                        
     /// combine: combineLatest__next(1A)
     /// combine: combineLatest__next(2A)
     /// combine: combineLatest__next(2B)
     /// combine: combineLatest__next(3B)
     /// combine: combineLatest__next(3C)
     /// combine: combineLatest__next(4C)
     /// combine: combineLatest__next(4D)
     /// combine: combineLatest__next(5D)
     /// combine: combineLatest__completed
     print(text)
}.disposed(by: disposeBag)

Observable<String>.combineLatest(observable6, observable5) { (e1:String, e2:Int) -> String in
     return "\(e1)\(e2)"
}.subscribe { event in
     let text = "combine: combineLatest__\(event)"
     self.showText(text: text)
                                                                        
     /// combine: combineLatest__next(A1)
     /// combine: combineLatest__next(B1)
     /// combine: combineLatest__next(B2)
     /// combine: combineLatest__next(C2)
     /// combine: combineLatest__next(C3)
     /// combine: combineLatest__next(D3)
     /// combine: combineLatest__next(D4)
     /// combine: combineLatest__next(D5)
     /// combine: combineLatest__completed
     print(text)
}.disposed(by: disposeBag)

Error错误处理相关

catch

在收到序列的异常事件时，通过返回另一个序列来持续发送非error事件。

Observable<UInt8>.create { observer in
     observer.onNext(0)
     observer.onError(NSError(domain: "error", code: 110, userInfo: nil))
     return Disposables.create()
}.catch { error -> Observable<UInt8> in
     let text = "combine: catch error__\(error)"
     self.showText(text: text)
                                  
     /// combine: catch error__Error Domain=error Code=110 "(null)"
     print(text)
     return Observable<UInt8>.of(1, 2)
}.subscribe { event in
     let text = "combine: catch__\(event)"
     self.showText(text: text)
                                              
     /// combine: catch__next(0)
     /// combine: catch__next(1)
     /// combine: catch__next(2)
     print(text)
}.disposed(by: disposeBag)

retry

出现错误事件后，重新发送所有事件信息。

Observable<UInt8>.create { observer in
      observer.onNext(0)
      observer.onError(NSError(domain: "error", code: 110, userInfo: nil))
      return Disposables.create()
}.retry(3).subscribe { event in
      let text = "combine: retry__\(event)"
      self.showText(text: text)
                                               
      /// combine: retry__next(0)
      /// combine: retry__next(0)
      /// combine: retry__next(0)
      /// combine: retry__error(Error Domain=error Code=110 "(null)")
      print(text)
}.disposed(by: disposeBag)

PracticalOperation实用操作相关

delay

延迟发射事件。

let text = "practicalOperation: delay__start time: \(Date())"
showText(text: text)

/// practicalOperation: delay__start time: 2021-03-14 09:38:36 +0000
print(text)
Observable<Int>.of(1, 2).delay(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe { event in
     if event.isCompleted {
         let text = "practicalOperation: delay__end time: \(Date())"
         self.showText(text: text)
                                                                                                            
         /// practicalOperation: delay__end time: 2021-03-14 09:38:37 +0000
         print(text)
      }
      let text = "practicalOperation: delay__\(event)"
      self.showText(text: text)
                                                                                                        
      /// practicalOperation: delay__next(1)
      /// practicalOperation: delay__next(2)
      /// practicalOperation: delay__completed
      print(text)
}.disposed(by: disposeBag)

do

在一个序列的每个事件执行之前添加一个执行动作。

Observable<Int>.of(1, 2, 3).do { _ in
     let text = "practicalOperation: do__previous next"
     self.showText(text: text)
     print(text)
} onError: { _ in
     let text = "practicalOperation: do__previous error"
     self.showText(text: text)
     print(text)
} onCompleted: {
    let text = "practicalOperation: do__previous complete"
    self.showText(text: text)
    print(text)
}.subscribe { event in
    let text = "practicalOperation: do__\(event)"
    self.showText(text: text)
             
    /// practicalOperation: do__previous next
    /// practicalOperation: do__next(1)
    /// practicalOperation: do__previous next
    /// practicalOperation: do__next(2)
    /// practicalOperation: do__previous next
    /// practicalOperation: do__next(3)
    /// practicalOperation: do__previous complete
    /// practicalOperation: do__completed
    print(text)
}.disposed(by: disposeBag)

observeOn

observer在指定scheduler中观察序列事件。

Observable<Int>.of(1).observe(on: ConcurrentDispatchQueueScheduler(queue: DispatchQueue(label: "test"))).subscribe { event in
     let text = "practicalOperation: observeOn__isMainThread: \(Thread.current.isMainThread)  \(event)"
     DispatchQueue.main.async {
          self.showText(text: text)
     }
                                                                                                                    
     /// practicalOperation: observeOn__isMainThread: false  next(1)
     /// practicalOperation: observeOn__isMainThread: false  completed
     print(text)
}.disposed(by: disposeBag)

subscribeOn

在指定的scheduler中操作，参考observeOn。

Observable<Int>.of(1).subscribe(on: MainScheduler.instance).subscribe { event in
     let text = "practicalOperation: subscribeOn__isMainThread: \(Thread.current.isMainThread)  \(event)"
     self.showText(text: text)
                                                                       
     /// practicalOperation: subscribeOn__isMainThread: true  next(1)
     /// practicalOperation: subscribeOn__isMainThread: true  completed
     print(text)
}.disposed(by: disposeBag)

timeout

一个序列在指定时间内未发射完成所有事件，那么将会进入.onError。

Observable<Int>.of(1).delay(RxTimeInterval.seconds(2), scheduler: MainScheduler.instance).timeout(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe { event in
     let text = "practicalOperation: timeout__\(event)"
     self.showText(text: text)

     /// practicalOperation: timeout__error(Sequence timeout.)
     print(text)
}.disposed(by: disposeBag)

ifEmpty

如果是序列中没有任何item，那么给定一个default。

Observable<Int>.empty().ifEmpty(default: 0).subscribe { event in
     let text = "practicalOperation: defaultIfEmpty__\(event)"
     self.showText(text: text)
                                                       
     /// practicalOperation: defaultIfEmpty__next(0)
     /// practicalOperation: defaultIfEmpty__completed
     print(text)
}.disposed(by: disposeBag)

skipUntil

丢弃掉第一个序列的所有items，直到第二个序列的item出现。

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()
sourceSequence.skip(until: referenceSequence).subscribe { event in
    let text = "practicalOperation: skipUntil__\(event)"
    self.showText(text: text)
                                                         
    /// practicalOperation: skipUntil__next(4)
    /// practicalOperation: skipUntil__next(5)
    /// practicalOperation: skipUntil__completed
    print(text)
}.disposed(by: disposeBag)
sourceSequence.onNext("1")
sourceSequence.onNext("2")
sourceSequence.onNext("3")
referenceSequence.onNext("1")
sourceSequence.onNext("4")
sourceSequence.onNext("5")

/// 必须调用onCompleted, 不然返回, 所在控制器, 不会被释放
sourceSequence.onCompleted()

skipWhile

丢弃掉所有的items，直到满足某个不满足条件的item出现。

Observable<String>.of("AD", "BD", "CD").skip { element -> Bool in
     return element.contains("A")
}.subscribe { event in
     let text = "practicalOperation: skipWhile__\(event)"
     self.showText(text: text)
                                                          
     /// practicalOperation: skipWhile__next(BD)
     /// practicalOperation: skipWhile__next(CD)
     /// practicalOperation: skipWhile__completed
     print(text)
}.disposed(by: disposeBag)

takeUntil

取得第一个序列所有items，直到第二个序列发射item或者终止。

Observable<Int>.of(1, 2, 3, 4, 5, 6, 7, 8, 9).take(until: { element -> Bool in
     return element > 4
}).subscribe { event in
     let text = "practicalOperation: takeUntil__\(event)"
     self.showText(text: text)
                                                                        
     /// practicalOperation: takeUntil__next(1)
     /// practicalOperation: takeUntil__next(2)
     /// practicalOperation: takeUntil__next(3)
     /// practicalOperation: takeUntil__next(4)
     /// practicalOperation: takeUntil__completed
     print(text)
}.disposed(by: disposeBag)

takeWhile

取得第一个序列的所有items，直到出现不满足条件的item （请仔细体会与skipWhile的不同之处）。

Observable<Int>.of(1, 2, 3, 4, 5, 6, 7, 8, 9).take(while: { element -> Bool in
     return element < 5
}).subscribe { event in
     let text = "practicalOperation: takeWhile__\(event)"
     self.showText(text: text)
                                                                        
     /// practicalOperation: takeWhile__next(1)
     /// practicalOperation: takeWhile__next(2)
     /// practicalOperation: takeWhile__next(3)
     /// practicalOperation: takeWhile__next(4)
     /// practicalOperation: takeWhile__completed
     print(text)
}.disposed(by: disposeBag)

Observer观察者&Observable被观察者相关

AnyObserver

let observer1 = AnyObserver<Int> { event in
                     let text = "observer: anyObserver__\(event)"
                     self.showText(text: text)
                                  
                     /// observer: anyObserver__next(1)
                     /// observer: anyObserver__next(2)
                     /// observer: anyObserver__next(3)
                     /// observer: anyObserver__completed
                     print(text)
                 }
Observable<Int>.of(1, 2, 3).subscribe(observer1).disposed(by: disposeBag)

AsyncSubject

Subject在ReactiveX的一些实现中扮演了一种桥梁或者代理的角色，它既可以作为Observer也可以作为Observable来使用。
作为观察者来说它可以订阅一个或者多个可观察序列，作为可观察者来说它可以通过items的reemitting来观察，并且还可以发射新的items事件，我们将从如下四个Subject进行学习：
AsyncSubject仅仅只发送订阅之后的最后一个item以及.onCompleted，如果出现错误，那么仅仅将只发送.onError。

let asyncSubject = AsyncSubject<Int>()
asyncSubject.onNext(1)
asyncSubject.onNext(2)
asyncSubject.subscribe { event in
     let text = "observer: asyncSubject__\(event)"
     self.showText(text: text)
                        
     /// observer: asyncSubject__next(4)
     /// observer: asyncSubject__completed
     print(text)
}.disposed(by: disposeBag)
asyncSubject.onNext(3)
asyncSubject.onNext(4)

/// 没有调用onCompleted, 则收不到subscribe, 因为不知道哪个是最后一个
asyncSubject.onCompleted()

ReplaySubject

订阅ReplaySubject的时候，可以接收到订阅他之后的事件，但也可以接受订阅他之前发出的事件，接受几个事件取决与bufferSize的大小。
createUnbounded()表示接受所有事件。
create(bufferSize: 4) 表示可接受到的订阅他之前的事件的个数，但是订阅他之后的事件一定会触发。

let replaySubject = ReplaySubject<Int>.createUnbounded()
replaySubject.onNext(1)
replaySubject.onNext(2)
replaySubject.subscribe { event in
      let text = "observer: replaySubject__\(event)"
      self.showText(text: text)
                         
      /// observer: replaySubject__next(1)
      /// observer: replaySubject__next(2)
      /// observer: replaySubject__next(3)
      /// observer: replaySubject__completed
      print(text)
}.disposed(by: disposeBag)
replaySubject.onNext(3)
replaySubject.onCompleted()

let replaySubject1 = ReplaySubject<Int>.create(bufferSize: 1)
replaySubject1.onNext(1)
replaySubject1.onNext(2)
replaySubject1.subscribe { event in
      let text = "observer: replaySubject__\(event)"
      self.showText(text: text)
                          
      /// observer: replaySubject__next(2)
      /// observer: replaySubject__next(3)
      /// observer: replaySubject__completed
      print(text)
}.disposed(by: disposeBag)
replaySubject1.onNext(3)
replaySubject1.onCompleted()

PublishSubject

订阅PublishSubject的时候，只能接收到订阅他之后发生的事件。subject.onNext()发出onNext事件，对应的还有onError()和onCompleted()事件，可以把他看成一个bufferSize=0的ReplaySubject。

let publishSubject = PublishSubject<Int>()
publishSubject.onNext(1)
publishSubject.onNext(2)
publishSubject.subscribe { event in
       let text = "observer: publishSubject__\(event)"
       self.showText(text: text)
                          
       /// observer: publishSubject__next(3)
       /// observer: publishSubject__completed
       print(text)
}.disposed(by: disposeBag)
publishSubject.onNext(3)
publishSubject.onCompleted()

BehaviorSubject

订阅了BehaviorSubject，会接受到订阅之前的最后一个事件，订阅之后的事件一定会触发。

let behaviorSubject = BehaviorSubject<Int>(value: 0)
behaviorSubject.onNext(1)
behaviorSubject.onNext(2)
behaviorSubject.subscribe { event in
        let text = "observer: behaviorSubject__\(event)"
        self.showText(text: text)
                           
        /// observer: behaviorSubject__next(2)
        /// observer: behaviorSubject__next(3)
        /// observer: behaviorSubject__completed
        print(text)
}.disposed(by: disposeBag)
behaviorSubject.onNext(3)
behaviorSubject.onCompleted()

BehaviorRelay

BehaviorReplay是Swift5 替换 Swift4 中的 Variable。
1 可以明确的是它不是subject类型，因为它只是一个可观察序列，但是它又包含subject对象（私有的BehaviorSubject）。
2 初始化的时候也需要一个初始值。
3 既然它不是一个订阅者，那么就不能发出onNext:、complete和error事件。
4 只能通过accept发出event。
总结：BehaviorRelay 跟 BehaviorSubject 很像，只是不是发出complete、error事件。

let behaviorRelay = BehaviorRelay(value: 0)
behaviorRelay.accept(1)
behaviorRelay.accept(2)
behaviorRelay.subscribe { event in
     let text = "observer: behaviorRelay__\(event)"
     self.showText(text: text)
                         
     /// observer: behaviorRelay__next(2)
     /// observer: behaviorRelay__next(3)
     /// observer: behaviorRelay__next(4)
     print(text)
}.disposed(by: disposeBag)
behaviorRelay.accept(3)
behaviorRelay.accept(4)

Driver

Driver从名字上可以理解为驱动，在功能上它类似被观察者（Observable），而它本身也可以与被观察者相互转换（Observable: asDriver, Driver: asObservable），它驱动着一个观察者，当它的事件流中有事件涌出时，被它驱动着的观察者就能进行相应的操作。一般我们会将一个Observable被观察者转换成Driver后再进行驱动操作。
Driver的drive方法与Observable的方法bindTo用法非常相似，事实上，它们的作用也是一样，说白了就是被观察者与观察者的绑定。
那为什么RxSwift的作者又搞出Driver这么个东西来呢？
比较与Observable，Driver有以下的特性：
1 它不会发射出错误(Error)事件。
2 对它的观察订阅是发生在主线程(UI线程)的。
3 自带shareReplayLatestWhileConnected。

let categorysDriver = categorysRelay.map { models -> [ServiceCategoryItemModel] in
     return [ServiceCategoryItemModel(model: Void(), items: models)]
}.asDriver(onErrorJustReturn: []).drive(headerView.categorysRelay).disposed(by: disposeBag)

Scheduler调度器相关

对于Scheduler来说，我们需要了解Concurrent（并行）、Serial（串行）Scheduler就可以了。

Observable<Int>.of(1, 2, 3).observe(on: SerialDispatchQueueScheduler(internalSerialQueueName: "serialDispatchQueue")).map { element -> Int in
     let text = "scheduler: map --> Main Thread: \(Thread.current.isMainThread) element__\(element)"
     DispatchQueue.main.async {
            self.showText(text: text)
     }
                                                                                                                           
     /// scheduler: map --> Main Thread: false element__1
     /// scheduler: map --> Main Thread: false element__2
     /// scheduler: map --> Main Thread: false element__3
     print(text)
     return element * 2
  
/// shareReplay(1)或shareReplayLatestWhileConnected，以防止以后被观察者被多次订阅观察后，map中的语句会多次调用：
}.share(replay: 1).subscribe(on: MainScheduler.instance).observe(on: MainScheduler.instance).subscribe { event in
     let text = "scheduler: subscribe --> Main Thread: \(Thread.current.isMainThread) event__\(event)"
     self.showText(text: text)
                                                                                                                          
     /// scheduler: subscribe --> Main Thread: true event__next(2)
     /// scheduler: subscribe --> Main Thread: true event__next(4)
     /// scheduler: subscribe --> Main Thread: true event__next(6)
     /// scheduler: subscribe --> Main Thread: true event__completed
     print(text)
}.disposed(by: disposeBag)

Observable<Int>.of(4, 5, 6).observe(on: ConcurrentDispatchQueueScheduler(queue: DispatchQueue(label: "concurrentDispatchQueue"))).map { element -> Int in
     let text = "scheduler: map --> Main Thread: \(Thread.current.isMainThread) element__\(element)"
     DispatchQueue.main.async {
          self.showText(text: text)
     }
                                                                                                                                       
     /// scheduler: map --> Main Thread: false element__4
     /// scheduler: map --> Main Thread: false element__5
     /// scheduler: map --> Main Thread: false element__6
     print(text)
     return element * 2
                                                                     
/// shareReplay(1)或shareReplayLatestWhileConnected，以防止以后被观察者被多次订阅观察后，map中的语句会多次调用：
}.share(replay: 1).subscribe(on: MainScheduler.instance).observe(on: MainScheduler.instance).subscribe { event in
     let text = "scheduler: subscribe --> Main Thread: \(Thread.current.isMainThread) event__\(event)"
     self.showText(text: text)
                                                                                                                                      
     /// scheduler: subscribe --> Main Thread: true event__next(8)
     /// scheduler: subscribe --> Main Thread: true event__next(10)
     /// scheduler: subscribe --> Main Thread: true event__next(12)
     /// scheduler: subscribe --> Main Thread: true event__completed
     print(text)
}.disposed(by: disposeBag)

Observable<Int>.of(7, 8, 9).observe(on: ConcurrentMainScheduler.instance).map { element -> Int in
     let text = "scheduler: map --> Main Thread: \(Thread.current.isMainThread) element__\(element)"
     DispatchQueue.main.async {
          self.showText(text: text)
     }
        
     /// scheduler: map --> Main Thread: true element__7
     /// scheduler: map --> Main Thread: true element__8
     /// scheduler: map --> Main Thread: true element__9
     print(text)
     return element * 2
                                                                               
/// shareReplay(1)或shareReplayLatestWhileConnected，以防止以后被观察者被多次订阅观察后，map中的语句会多次调用：
}.share(replay: 1).subscribe(on: MainScheduler.instance).observe(on: MainScheduler.instance).subscribe { event in
     let text = "scheduler: subscribe --> Main Thread: \(Thread.current.isMainThread) event__\(event)"
     self.showText(text: text)
                                                                              
     /// scheduler: subscribe --> Main Thread: true event__next(14)
     /// scheduler: subscribe --> Main Thread: true event__next(16)
     /// scheduler: subscribe --> Main Thread: true event__next(18)
     /// scheduler: subscribe --> Main Thread: true event__completed
     print(text)
}.disposed(by: disposeBag)

Disposable

订阅了一个可观察序列，如果有特殊需求需要提前取消订阅时使用。也就是说Disposable是用来取消订阅的一个工具，通过Disposables工具创建。

let dis1 = Disposables.create()
let dis2 = Disposables.create {
    print("在Dispose之前所做一些工作")
}
let _ = Disposables.create([dis1, dis2])

/// dispose：通过.dispose()取消或者添加到disposeBag（可以将它看成一个非ARC机制下的AutoReleasePool)
let disposable = Observable<Int>.of(1, 2, 3).subscribe { event in
                      print(event)
                 }
disposable.dispose()

/// 或者
disposable.disposed(by: disposeBag)

常用方法

监听点击事件和输入事件

closeButton.rx.tap.subscribe(onNext: {[weak self] _ in
     self?.closeClosure?()
}).disposed(by: disposeBag)

phoneTextField.rx.text.orEmpty.subscribe(onNext: {[weak self] (text) in
     self?.textDidChange()
     self?.textDidChangeClosure?(text)
}).disposed(by: disposeBag)

phoneTextField.rx.controlEvent(.editingDidBegin).subscribe(onNext: {[weak self] _ in
     self?.editingDidBeginClosure?()
}).disposed(by: disposeBag)

phoneTextField.rx.controlEvent(.editingDidEnd).subscribe(onNext: {[weak self] _ in
     self?.editingDidEndClosure?()
}).disposed(by: disposeBag)

代替通知

// MARK: public
extension NotificationManager {

    /**
     注册通知
     */
    static func registNotification(_ name:NSNotification.Name, disposeBag:DisposeBag, callBack:@escaping ((_ noti:Notification) -> ())) {
        NotificationCenter.default.rx.notification(name).subscribe(onNext: {(noti) in
            callBack(noti)
        }).disposed(by: disposeBag)
    }

    /**
     发送通知
     */
    static func post(name aName: NSNotification.Name, object anObject: Any? = nil, userInfo aUserInfo: [AnyHashable : Any]? = nil) {
        NotificationCenter.default.post(name: aName, object: anObject, userInfo: aUserInfo)
    }
}

代替观察者

/// 只发出与下一个间隔超过0.25秒的元素
scrollView.rx.observe(String.self, "contentSize").debounce(RxTimeInterval.milliseconds(250), scheduler: MainScheduler.instance).subscribe(onNext: {[weak self] _ in
    self?.contentSizeClosure?(self?.scrollView.contentSize ?? .zero)
}).disposed(by: self.disposeBag)

信号合并

/// 将下拉刷新/上拉加载更多的信号和请求参数信号合并, 确定最后发起请求的请求参数
Observable<NewsListParams>.combineLatest(input.paramsRelay, input.reloadDataSubject) {[weak self] (params:NewsListParams, reloadData:Bool) -> NewsListParams in
     params.page = self?.caculatePage(reloadData) ?? 1
     params.row = self?.pageSize ?? 10
     return params
 }.subscribe(onNext: {[weak self] params in
     self?.showHUD()
     NewsManager.requestNewsListData(params: params) { datas in
           HUD.dismiss()
           self?.handleData(datas?.rows ?? [], scrollView: self?.tableView ?? BaseTableView(), isFirstFetch: params.page == 1, totalCount: datas?.total ?? 0)
     }
}).disposed(by: disposeBag)

数据绑定

let categorysDriver = categorysRelay.map { models -> [ServiceCategoryItemModel] in
     return [ServiceCategoryItemModel(model: Void(), items: models)]
}.asDriver(onErrorJustReturn: []).drive(headerView.categorysRelay).disposed(by: disposeBag)

代替定时器

/// Interval：创建一个可观察序列，以特定的时间间隔释放一系列整数(E -> Int/NSInteger)
Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).take(3).subscribe { event in
     let text = "operator: interval__\(event)"
     self.showText(text: text)

     /// operator: interval__next(0)
     /// operator: interval__next(1)
     /// operator: interval__next(2)
     /// operator: interval__completed
     print(text)
}.disposed(by: disposeBag)

/// Timer：在指定的时间后，发送一个特定的Item (E -> Int/NSInteger)，请注意这里与Interval的区别（Interval是发送一系列特定Item，而Timer只会发送一个）
Observable<Int>.timer(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe { event in
     let text = "operator: timer__\(event)"
     self.showText(text: text)

     // operator: timer__next(0)
     /// operator: timer__completed
     print(text)
}.disposed(by: disposeBag)