函数响应式编程FRP

之前学习过Objective-C下的FRP第三方开源框架ReactiveCocoa，部分学习笔记：
https://blog.csdn.net/wf96390/article/details/50933421
它可以在MVVM架构模式中充当着View（视图）层与ViewModel（视图模型）层之间的Binder（绑定者）角色，实现两个层之间的同步更新。在ReactiveCocoa的世界中，数据与属性的改变、视图的操作反馈、方法的调用等都可以被监听并抽象转换成事件流，封装在Signal（信号）中，我们通过对Signal的Subscribe（订阅）就能获取到其中的事件流，并进行相应的操作。

函数式编程 Functional programming

总结一下函数式编程具有以下几个特点：

• 函数是“第一等公民”
• 闭包和高阶函数
• 只用“表达式”，不用“语句”，不改变状态，没有副作用
• 柯里化

Haskell是一种标准化的、通用纯函数式编程语言，有非限定性语义和强静态类型。

响应式编程

var b = 1
let c = 2
let a = b  + c  // a = 3

// 改变 b 的值，a的值会变吗？
b = 3
print(a)

在命令式编程环境中， a = b + c 表示将表达式的结果赋给 a，而之后改变 b 或 c的值不会影响 a。
但在响应式编程中，a的值会随着 b或 c的更新而更新。

RxSwift

RxSwift是Reactive下的一个分支，Reactive涵盖了很多语言。

RxSwift为ReactiveX（Reactive Extensions）旗下的Swift语言库，提供了Swift平台上进行响应式编程的解决方案。Rx的重要角色为Observable（被观察者）和Observer（观察者），Observable类似于ReactiveCocoa中的Signal，里面装有事件流，供Observer订阅。事件流在Rx中与ReactiveCocoa一样具有三类：Next、Error、Completed，代表着继续事件、错误事件、完成事件。我们在使用RxSwift进行iOS开发时，通常会引入另外一个库：RxCocoa，这个库将UIKit以及Foundation框架中许多成员，如视图(View)、控制事件(Control Event)、键值观察（KVO）、通知（Notification）等等进行与RxSwift接入的扩展，将Rx与iOS API无缝连接。

Observable 和 Observer

Observable指的是可被观察者或者事件源。Observable是一种像信号一样的序列，可以理解为一串数组，但是需要有人监听这些信号。ObservableType协议中定义的subscribe函数就是一种简便的订阅信号的一种方法。这里的subscribe函数就是把消息发给观察者。

实现了 ObservableType 协议, ObservableType 协议中定义了很多方法。

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

    public func asObservable() -> RxSwift.Observable<RxSwift.Observable<Element>.E>
}

Observer : 观察者，观察者需要订阅Observable，才能受到其发出的事件

subscribe :用于订阅sequence发出的事件，相当于Swift序列中的遍历操作（makeIterator）

subscribe(_:)方法接受事件参数，下面这个例子接受元素参数
someObservable.subscribe(
    onNext: { print("Element:", $0) },
    onError: { print("Error:", $0) },
    onCompleted: { print("Completed") },
    onDisposed: { print("Disposed") }
)

Dispose

在使用subscribe订阅一个可观察序列时，会返回一个Disposable类型的对象。这里的Disposable是一个协议，只定义了一个方法：

public protocol Disposable {
    /// Dispose resource.
    func dispose()
}

DisposeBag : 类似于 iOS 中的 ARC，会在适当的时候销毁观察者，自动去释放资源。需要调用 disposed 方法，加入到 disposed 中。如果未添加 Xcode 会提示报错。

let disposeBag = DisposeBag()

extension Disposable {

    /// Adds `self` to `bag`
    ///
    /// - parameter bag: `DisposeBag` to add `self` to.
    public func disposed(by bag: RxSwift.DisposeBag)
}

如果需要手动销毁观察者直接调用dispose()方法

Observable.empty().subscribe { (str: Event) in 
    print(“empty:”, str) 
}.dispose()

常见用法

Observable 类似 swift 中的序列，可以用数据流图表示
https://rxmarbles.com/

创建Observable

• never

构建一个从不终止和发出任何事件的队列

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

• empty

构建一个空的Observable队列，只发出完成事件

 Observable<Int>.empty()
     .subscribe { event in
         print(event)
     }
     .disposed(by: disposeBag)

• just

构建一个只有一个元素的Observable队列

Observable.just("123")
   .subscribe { event in
       print(event)
    }
    .disposed(by: disposeBag)

• of

构建一个拥有固定数量元素的Observable序列

Observable.of("A", "B", "C", "D")
    .subscribe(onNext: { element in
        print(element)
    })
    .disposed(by: disposeBag)

• from

从序列中创建可观察到的序列，如数组、字典或集合。

Observable.from(["🐶", "🐱", "🐭", "🐹"])
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• create

构建一个自定义的可观察序列

  let myJust = { (element: String) -> Observable<String> in
      return Observable.create { observer in
          observer.on(.next(element))
          observer.on(.completed)
          return Disposables.create()
      }
  }
 myJust("🔴")
     .subscribe { print($0) }
     .disposed(by: disposeBag)

• range

创建一个可观察序列，该序列释放一系列连续整数，然后终止

 Observable.range(start: 1, count: 10)
    .subscribe { print($0) }
    .disposed(by: disposeBag)

• repeatElement

创建一个可观察到的序列，它无限地释放给定的元素

Observable.repeatElement("123")
     .take(3)
     .subscribe(onNext: { print($0) })
     .disposed(by: disposeBag)

• generate

创建一个可观察的序列，只要所提供的条件求值为true，就生成值。

 Observable.generate(
        initialState: 0,
        condition: { $0 < 3 },
        iterate: { $0 + 1 }
    )
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• error

创建一个没有任何元素的可观察序列，并立即以错误结束。

 Observable<Int>.error(TestError.test)
    .subscribe { print($0) }
    .disposed(by: disposeBag)

创建Subject

Subjet 类似于ReactiveCocoa中的热信号，可以主动发送信号，也可以被监听，一个Subject既可以充当Obserable也可以充当Observer，即它可以发出事件，也可以监听事件。
在被订阅以后，订阅者可以监听到订阅之后发生的事件，像收音机发送广播一样，打开收音机以后可以听到当前时间以及之后的广播，之前未收听到能否接收的根据Subject类型判断。
几种 Subject 都是 RxSwift.Observable<Element> 的子类。

• PublishSubject

采用PublishSubject订阅事件的时候，只能接收到订阅他之后发生的事件，完全和收音机听广播一样。

let publicSubject = PublishSubject<String>()
publicSubject.onNext("A")
publicSubject.onNext("B")
publicSubject.subscribe(onNext: {print("publicSubject:",$0)}).addDisposableTo(bag)
publicSubject.onNext("C")
publicSubject.onNext("D")

/**
 publicSubject: C
 publicSubject: D
 **/

• ReplaySubject

可以接收到订阅他之后以及之前发生的事件, 对于要接受几个以前的事件取决于bufferSize设置的大小。

let replaySubject = ReplaySubject<String>.create(bufferSize: 2)
// 如果需要接受全部的事件，则可以使用
// let replaySubject = ReplaySubject<String>.createUnbounded()

replaySubject.onNext("A")
replaySubject.onNext("B")
replaySubject.onNext("C")
replaySubject.onNext("D")
replaySubject.subscribe(onNext: {print("replaySubject:",$0)}).addDisposableTo(bag)
replaySubject.onNext("E")
replaySubject.onNext("F")
replaySubject.onNext("G")
/**
 replaySubject: C
 replaySubject: D
 replaySubject: E
 replaySubject: F
 replaySubject: G
 **/

• BehaviorSubject

采用BehaviorSubject订阅事件，会接收到订阅之前的最后一个事件以及订阅之后所有事件。如果在 BehaviorSubject 创建后马上监听，会接收到初始值和订阅之后的事件。

let behavior = BehaviorSubject(value: "behavior")
behavior.onNext("A")
behavior.onNext("B")
behavior.subscribe(onNext: {print("behavior:",$0)}).addDisposableTo(bag)
behavior.onNext("C")
behavior.onNext("D")

/**
 behavior: B
 behavior: C
 behavior: D
 **/

• Variable

Variable 是对 BehaviorSubject 的一个包装。BehaviorSubject 不会因为错误而终止, 但是 variable 会被释放。
Variable当成Obserable, 让订阅者进行订阅时, 需要asObserable转成Obserable；

Variable发出事件, 直接修改对象的value即可;

当事件结束时,Variable会自动发出completed事件

This concept will be deprecated from RxSwift but offical migration path hasn't been decided yet.
https://github.com/ReactiveX/RxSwift/issues/1501

Current recommended replacement for this API is RxCocoa.BehaviorRelay because:

• Variable isn't a standard cross platform concept, hence it's out of place in RxSwift target.
• It doesn't have a counterpart for handling events (PublishRelay). It models state only.
• It doesn't have a consistent naming with *Relay or other Rx concepts.
• It has an inconsistent memory management model compared to other parts of RxSwift (completes on deinit).

let variable = Variable("variable")

variable.value = "A"
variable.value = "B"
variable.asObservable().subscribe { (event:Event<String>) in
    print("variable:",event)
    }.addDisposableTo(bag)
variable.value = "C"
variable.value = "D"

/**
 variable: next(B)
 variable: next(C)
 variable: next(D)
 variable: completed
 **/

组合操作

可以将多个Observables组合成单个的Observable

• startWith

在Observable源发出元素前，发出特定的元素

Observable.of("2", "3")
    .startWith("1️⃣")
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• merge

将源可观察序列的元素组合成一个新的可观察序列

let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()
    
Observable.of(subject1, subject2)
    .merge()
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
    
subject1.onNext("🅰️")
subject1.onNext("🅱️")
subject2.onNext("①")
subject2.onNext("②")
subject1.onNext("🆎")
subject2.onNext("③")

• zip

将可观察序列合并成一个新的可观察序列

let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()
    
Observable.zip(stringSubject, intSubject) { stringElement, intElement in
    "\(stringElement) \(intElement)"
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
    
stringSubject.onNext("🅰️")
stringSubject.onNext("🅱️")
    
intSubject.onNext(1)
intSubject.onNext(2)

stringSubject.onNext("🆎")
intSubject.onNext(3)

• combineLatest

将可观察队列合并成单个新的可观察队列，当多个源至少有一个的时候发出信号

let stringSubject = PublishSubject<String>()
let intSubject = PublishSubject<Int>()

Observable.combineLatest(stringSubject, intSubject) { stringElement, intElement in
        "\(stringElement) \(intElement)"
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

stringSubject.onNext("🅰️")

stringSubject.onNext("🅱️")
intSubject.onNext(1)

intSubject.onNext(2)

stringSubject.onNext("🆎")

当然可以将数组作为参数

let stringObservable = Observable.just("❤️")
let fruitObservable = Observable.from(["🍎", "🍐", "🍊"])
let animalObservable = Observable.of("🐶", "🐱", "🐭", "🐹")

Observable.combineLatest([stringObservable, fruitObservable, animalObservable]) {
        "\($0[0]) \($0[1]) \($0[2])"
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

如果你采用zip组合，那必须两个label都有值改变才会触发，如果使用combineLatest，则其中一个改变就会出发。

转换操作

转换可观察队列发出的Next事件里元素的操作

• map

将转换闭包应用于可观察序列发出的元素，并返回已转换元素的一个新的可观察序列。

let disposeBag = DisposeBag()
Observable.of(1, 2, 3)
    .map { $0 * $0 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• flatMap

将Observable队列发出的元素转换成一个新的可观察队列，并将两者的输出组合成一个新的可观察队列，意思就是说会监听原队列，也会监听你闭包中转换过的队列。

struct Player {
    var score: Variable<Int>
}

let 👦🏻 = Player(score: Variable(80))
let 👧🏼 = Player(score: Variable(90))

let player = Variable(👦🏻)

player.asObservable()
    .flatMap { $0.score.asObservable() } // Change flatMap to flatMapLatest and observe change in printed output
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

👦🏻.score.value = 85  // 会有响应

player.value = 👧🏼    // 这边也会有响应  监听了两个组合队列

👦🏻.score.value = 95 // Will be printed when using flatMap, but will not be printed when using flatMapLatest

👧🏼.score.value = 100

• scan

以初始值开始，然后将一个累加器闭包应用于可观察序列所发射的每个元素，并将每个中间结果作为一个元素可观察序列返回。看例子

Observable.of(10, 100, 1000)
    .scan(1) { aggregateValue, newValue in    // aggregateValue 之前返回的值
        aggregateValue + newValue
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)    //  11,111,1111

过滤操作

• filter

类似于 swift 语法中的filter，筛选出符合条件的信号

Observable.of(
    "🐱", "🐰", "🐶",
    "🐸", "🐱", "🐰",
    "🐹", "🐸", "🐱")
    .filter {
        $0 == "🐱"
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

• distinctUntilChanged
  直到信号变化时，才发出信号，比如使用 textfield 只有内容变化时才发出信号

Observable.of("🐱", "🐷", "🐱", "🐱", "🐱", "🐵", "🐱")
    .distinctUntilChanged()
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)    //🐱🐷🐱🐵🐱

• elementAt
  获取序列中的第某个信号

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .elementAt(3)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)   //   🐸

• single

序列发出的第一个元素（或满足条件的第一个元素）。如果序列中有多个满足条件的或者不存在满足条件的，则抛出一个错误。

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .single()
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

增加判断条件

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .single { $0 == "🐸" }
    .subscribe { print($0) }
    .disposed(by: disposeBag)   //  next(🐸)   completed

Observable.of("🐱", "🐰", "🐶", "🐱", "🐰", "🐶")
    .single { $0 == "🐰" }
    .subscribe { print($0) }
    .disposed(by: disposeBag)     // next(🐰) error(Sequence contains more than one element.)

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .single { $0 == "🔵" }
    .subscribe { print($0) }
    .disposed(by: disposeBag)  //  error(Sequence doesn't contain any elements.)

• take

从序列的开头获取指定数量的信号

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .take(3)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)  //     🐱🐰🐶

• takeLast

从序列的结尾获取指定数量的元素。

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .takeLast(3)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)  //   🐸🐷🐵

• takeWhile

在指定的条件满足时，则从可观察序列的开头发出元素。

Observable.of(1, 2, 3, 4, 5, 6)
    .takeWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)  // 1 2 3

• takeUntil

从源序列中获取元素，直到引用可观察序列发出元素结束

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()

sourceSequence
    .takeUntil(referenceSequence)
    .subscribe { print($0) }
    .disposed(by: disposeBag)

sourceSequence.onNext("🐱")
sourceSequence.onNext("🐰")
sourceSequence.onNext("🐶")

referenceSequence.onNext("🔴")

sourceSequence.onNext("🐸")
sourceSequence.onNext("🐷")
sourceSequence.onNext("🐵")   // next(🐱)  next(🐰)  next(🐶)  completed

• skip

从序列的开头跳过指定数量的元素。

Observable.of("🐱", "🐰", "🐶", "🐸", "🐷", "🐵")
    .skip(2)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)  // 🐶🐸🐷🐵

• skipWhile

从序列的开头跳过满足条件的元素。

Observable.of(1, 2, 3, 4, 5, 6)
    .skipWhile { $0 < 4 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)   // 4 5 6

• skipUntil

从源序列中跳过元素，直到引用可观察序列发出元素。

let sourceSequence = PublishSubject<String>()
let referenceSequence = PublishSubject<String>()

sourceSequence
    .skipUntil(referenceSequence)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

sourceSequence.onNext("🐱")
sourceSequence.onNext("🐰")
sourceSequence.onNext("🐶")

referenceSequence.onNext("🔴")

sourceSequence.onNext("🐸")
sourceSequence.onNext("🐷")
sourceSequence.onNext("🐵")  // 🐸🐷🐵

MVVM-C

参见下一篇文章

https://www.jianshu.com/p/d1386dc7617d

CSDN博客地址：
https://blog.csdn.net/wf96390/article/details/88086808