响应式编程的复杂度和简化
Posted 阿里巴巴淘系技术团队官网博客
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了响应式编程的复杂度和简化相关的知识，希望对你有一定的参考价值。
响应式系统不是今天的主题，我们要讨论更具体的话题，即响应式代码的编写会有哪些复杂度，应该如何简化。
什么是响应式编程
什么是响应式编程，它是一种编程范式？还是一种设计模式？抑或是其他？响应式系统和响应式编程有什么关系？又比如，响应式编程它适用于什么场景？解决什么问题？
微软于2011年率先建设了.Net上的Rx库，以简化容易出错的异步和事件驱动编程，迈出了响应式编程的第一步，随后业界为许多编程语言提供了对应的实现。
.Net上的Rx库地址：https://docs.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh242985(v=vs.103)
什么是响应式，我们从一个例子开始，
在上面的表格中，建立了单元格之间的关系：`A1 = B1 + C1`，建立关系之后A1将响应任何对于B1和C1的变化，毫无疑问，这就是一种响应式行为。
我觉得这个例子很棒的地方在于，它显然很简单，同时又足够深刻，首先，它充分的体现了响应式的概念，其次，变化发生时，肯定触发了某些过程的执行，说明背后存在关系的建立和沿着关系传播的变化，再次，稍微深入一点看，B1和C1的变化可以是一系列的变化，可以很自然的引申到流的概念，最后，它有一个很高级的抽象，对使用方来说，整个过程是声明式的。
当然，举例子来说明一个概念的时候，本质上是用一个外延在解释一个概念的内涵，往往是会将内涵缩小的，所以我们可以尝试推广这个外延。列出这个例子的特征：
描述了一个单元格里的整数等于另外两个单元格里整数相加，当后者每次发生变化时，变化都会传播到第一个单元格，并进行求值。
关键词为整数、等于、相加、变化、每次、传播、求值。前三个关键词仅仅和例子相关，可以直接去掉。变化可以推广为数据，每次可以在逻辑上等价于流的概念，流可以有0个、1个或多个数据，传播可以推广为通信（在这个意义上，函数调用、RPC、socket、MQ都是通信），求值推广为执行一个过程。所以我们可以得出响应式编程的定义：
通过声明式的通信定义，将数据流与过程组合起来，从而实现数据驱动过程的一种复合编程范式。
时至今日，业界对于响应式的定义仍然是不统一的，因此这是我自己的理解。响应式的基础概念是数据流，理念是过程的执行是通过响应数据来驱动的，核心是构造数据和过程的响应关系，并且能够让数据沿着关系传播驱动过程，因此响应式编程本质上是一种对通信的抽象，说它是一种编程范式，是因为它提供一种对于数据与过程组合方式的看法，说它是复合范式而不是基本范式，是因为它不像OOP或者FP一样提供的是对于数据和过程的看法，而是以两者为基础，所以可以有对象响应式和函数响应式。
当我们基于响应式构建系统时，就是响应式系统，响应式系统的构建原则可以参考此处（地址：https://www.reactiveprinciples.org/patterns/communicate-facts.html），总的来说，系统会分割成一个一个的分区，分区内部对状态进行本地化，分区之间通过通信进行异步解耦，可以通过控制这个通信的过程，实现系统的弹性扩缩容和部分组件失败的回弹性。
响应式系统不是今天的主题，我们要讨论更具体的话题，即响应式代码的编写会有哪些复杂度，应该如何简化。
响应式编程的复杂度
响应式编程的复杂度来自于4个方面：
可以有0次、1次或多次数据产生，也就是数据流；除了数据之外，还有能够标识错误和完成（正常结束）；
数据流和数据流、数据流和过程的组合复杂度很高；
在上面的基础上，需要处理整个过程中线程切换、并发同步、数据缓冲等问题。
为了支持数据流的概念，可以产生0次、1次或多次数据产生，API设计需要把数据回调和结果回调分开，通常也会把错误回调和完成回调分开，这种接口被称为流式接口，一个标准的流式接口设计如下所示：
typealias Func = () -> ()
typealias OnData<Data> = (Data) -> ()
typealias OnError = (Error) -> ()
typealias OnComplete = Func
typealias StreamFunc<Data> = (@escaping OnData<Data>, @escaping OnError, @escaping OnComplete) -> ()显然，流式接口是普通异步接口将一次结果向多次结果的推广，这种推广同时也增加了逻辑的复杂度。
我们可以通过一个逻辑上简单的例子来看一下流式接口的使用过程，为了关注于核心的复杂度，只会体现前3个方面，一方面是由于加入第4点的话会导致代码过于冗长混淆关注点，另一方面相信各位对第4点本身的复杂度和它引起的众多问题已经非常熟悉了。
这个例子很简单，只有三步：
假设需要为一个店铺提供一个订单展示页面，这些订单来自两个不同的平台“鹅鹅鹅”和“鸭鸭鸭”，他们各自提供了查询的接口（listOrders，为了简单假设他们提供的模型和接口完全一致）；
订单列表需要展示用户的昵称等信息，需要通过对应平台的另外一个接口（queryUserInfo）查询；
由于SDK缓存、持久化、网络请求策略，数据无法一次性获取，这两个接口可能存在多次数据回调。
进一步简化问题，我们忽略变更处理、UI渲染和用户交互处理，仅仅考虑数据加载，这需要组合2个阶段的4次接口调用，先分别请求两个平台的订单，使用订单请求对应平台的userInfo，最后合并成完整数据：
// 数据怎么回调，什么情况结束，onError和onComplete分别在什么情况回调，保证有且仅有一次回调
func load(onData : OnData<[OrderObject]>?, onError : OnError?, onComplete : OnComplete?)
let orderServices = [OrderService("鹅鹅鹅"), OrderService("鸭鸭鸭")]
// 记录整体请求的完成状态
var listOrderFinish = false
var queryUserFinish = false
// 记录各个请求的结果
var listOrderResults = orderServices.map_ in false
var queryUserResults = [Bool]()
for (index, orderService) in orderServices.enumerated()
orderService.listOrders orders in
// 已结束不处理
if (listOrderFinish)
return;
let index = queryUserResults.count
queryUserResults[index] = false
if let userService = getUserService(site: orderService.site)
let userIds = orders.map order in
order.userId
userService.queryUserInfo(userIds: userIds) userInfoDict in
if (listOrderFinish && queryUserFinish)
return;
let orderObjects = orders.map order in
OrderObject(order: order, userInfo: userInfoDict[order.userId])
onData?(orderObjects)
onError: error in
// 如果是第一个错误，直接回调，同时标记为结束
if (!listOrderFinish || !queryUserFinish)
listOrderFinish = true
queryUserFinish = true
onError?(error)
onComplete:
// 外层结束，内层也结束，才是最终结束
if (!listOrderFinish || !queryUserFinish)
queryUserResults[index] = true
// 所有都结束，回调
if (listOrderFinish && !queryUserResults.contains(false))
listOrderFinish = true
onComplete?()
else
let orderObjects = orders.map order in
OrderObject(order: order)
onData?(orderObjects)
queryUserResults[index] = true
// 所有都结束，回调
if (listOrderFinish && !queryUserResults.contains(false))
listOrderFinish = true
onComplete?()
onError: error in
// 如果是第一个错误，直接回调，同时标记为结束
if (!listOrderFinish)
listOrderFinish = true
onError?(error)
onComplete:
// 注意，即使所有的请求都结束了，也不能回调结束，因为这里的结束只是代表Order请求结束，userInfo请求不一定结束
if (!listOrderFinish)
listOrderResults[index] = true
// 所有都结束，回调
if (!listOrderResults.contains(false))
listOrderFinish = true
在这个接口的实现中，数据回调最简单，在没有结束的情况下，多次回调的数据可以直接回调，问题是如何保证错误和完成有且仅有一次回调，且结果回调后不再回调数据，即：
什么时候回调错误？
什么时候回调完成？
如果我们认为一个接口出错，就回调错误，这是最简单的错误处理，只需要检查和设置结束状态，在没有结束时的第一个错误进行回调即可，注意，我们需要在userInfo的请求中也做类似的处理，并保证错误回调后不再执行任何回调。
完成的回调要比错误复杂的多，我们可以来思考一下：
首先，我们不能在listOrders的onComplete里面取回调完成，因为这里不能代表queryUserInfo这个接口也完成了；
其次，我们也不能简单的通过所有queryUserInfo都完成了就回调完成，因为listOrders在完成前仍然有可能返回新的订单数据。
也就是说，这里的完成需要在queryUserInfo进行判断，并且也需要考虑外层请求的完成情况，比普通异步接口的级联要多了两个维度。这仅仅是2种接口4次请求，在真实的编程中，接口数量会多得多，并且需要把第4点加进来，线程/队列、并发、同步、缓冲区，还要处理新数据推送响应，再考虑调试、监控、排查，复杂度显然会继续大幅增长，保证这个过程的正确性是一件痛苦的事情。
响应式编程的复杂度使用Rx/Combine简化响应式编程
为了解决这些问题，业界搞出了Reactive Streams规范（地址：https://www.reactive-streams.org/），也出现了若干的实现，都以工具库的形式提供，包括Rx系列、Reactor，以及苹果功能类似的Combine。作为一个ios开发，我对RxSwift和Combine比较了解，两者主要的区别在于Combine多了一个Subscription的抽象来协调Publisher和Subscriber之间的行为，尤其是Back Pressure相关的控制，但总的来说，都提供了对于异步数据流的抽象和组合能力，用法上也很类似，这里以RxSwift为例来重写上面的过程。
第一步，实现一个将流式函数转换成Observable的工具类，这个是通用的，非常直观：
func makeObservable<Data>(f : @escaping StreamFunc<Data>) -> Observable<Data>
Observable<Data>.create observer in
f data in
observer.onNext(data)
_: error in
observer.onError(error)
_:
observer.onCompleted()
return Disposables.create()
第二步，针对这个例子，将listOrder和queryUserInfo转换成StreamFunc形式，listOrder本来就是StreamFunc，对queryUserInfo进行偏应用也可以转换为StreamFunc形式，这是具体接口相关的：
func makeStreamFunc(orders : [Order], userInfoService : UserService?) -> StreamFunc<[OrderObject]>
if let userInfoService = userInfoService
// 核心是对queryUserInfo的userIds参数进行偏应用
let userInfoF : StreamFunc<[OrderObject]> = onData, onError, onComplete in
let userIds = orders.map$0.userId
userInfoService.queryUserInfo(userIds: userIds, onData: userInfoDict in
let orderObjects = orders.map order in
OrderObject(order: order, userInfo: userInfoDict[order.userId])
onData(orderObjects)
, onError: onError, onComplete: onComplete)
return userInfoF
else
return onData, onError, onComplete in
onData(orders.mapOrderObject(order: $0))
onComplete()
第三步，这样就可以将load方法简化为：
func rxLoad() -> Observable<[OrderObject]>
let orderService = [OrderService("鹅鹅鹅"), OrderService("鸭鸭鸭")]
// 通过map构造Observable，通过flatMap对listOrder和queryUserInfo进行复合
let observables = orderService.map orderService in
makeObservable(f: orderService.listOrders).flatMap (orders) -> Observable<[OrderObject]> in
let userLoadF = makeStreamFunc(orders: orders, userInfoService: getUserService(site: orderService.site))
return makeObservable(f: userLoadF)
// merge两个平台的Observable
return Observable.merge(observables)
可以看到，第一步是通用的，实际代码中只需要做第二步和第三步，这就对上面的接口进行了大量的简化，并且库以统一的方式处理掉了合并、级联、多数据返回的复杂逻辑，我们有相当的把握来保证正确性。当然，除了学习成本较高以外，也还是有缺点的，主要是使用方式仍然是异步形式，在部分环节仍然需要处理异步带来的复杂度：
// 使用方调用
rxLoad().subscribe orderObjects in
// onNext闭包中处理数据
onError: error in
// onError闭包中处理错误
onCompleted:
// onCompleted闭包中处理完成
onDisposed:
Rx确实大大简化了异步编程，但是还不够，因为它的使用仍然是异步形式。
使用AsyncSequence简化响应式编程
▐ 迭代器与序列
迭代器是很多语言都有的一个概念，一个迭代器的核心是next()函数，每次调用都会返回下一个数据，这些数据构成了一个序列（Sequence），迭代器也意味着序列可以被遍历。
▐ 异步序列
如果让迭代器的next()方法支持异步，就产生了异步序列。Swift对此提供了一个AsyncSequence的协议，并对它提供了语言级别的支持，使得开发者可以以同步的形式遍历一个异步序列：
for try await data in asyncDataList
print("async get data : + \\(data)")
实际上，Swift在Combine中支持了Publisher的同步遍历：
// Combine的同步调用
for try await data in publisher.values
print("async get publiser value \\(data)")
不过这个特性需要iOS15才能支持，如果说iOS13还可以展望的话，iOS15就是遥遥无期了。
▐ CPS变换
如果能将流式接口转换为异步序列，那么就可以实现响应式代码的同步编写，这个转换过程可以通过CPS变换实现。
CPS变换全称Continuation-Pass-Style，这个概念来自Lisp语系，是一种显式传递控制流的编程风格，其传递控制流的载体就是continuation。continuation可以理解为当前代码执行的后续，如果一个函数f有一个continuation参数，我们就可以把当前的continuation传递进去，当函数产生结果时，通过continuation回到函数f外，继续执行，这种函数调用方式成为call/cc（call with current continuation）。
这种变换，称为CPS变换。
作为一个类比，我觉得可以将continuation理解为return的在两个方面的推广形式，首先，continuation是first-class的，可以作为变量存储，可以作为函数的参数和返回值，其次，continuation可以多次使用，而return只能有一次。
▐ 响应式编程的同步形式
回头看最原始的代码，当我们调用orderService.listOrders时，传进去的callback，其实就相当于一个弱化版的continuation。这意味着，如果我们可以将使用continuation将数据表示为AsyncSequence，那么就可以将响应式代码写成同步形式，从而大幅简化响应式编程。
Swift提供了continuation的概念，提供了AsyncStream和AsyncThrowingStream来实现这个过程，对上节Rx的实现稍作改动即可。
第一步，实现一个将流式函数转换成AsyncThrowingStream的工具类，这个是通用的：
func makeSequence<Data>(f : StreamFunc<Data>) -> AsyncThrowingStream<Data, Error>
AsyncThrowingStream<Data, Error> continuation in
f data in
continuation.yield(data)
_: error in
continuation.finish(throwing: e)
_:
continuation.finish()
第二步，由于AsyncSequence还不支持merge，需要自己实现一个merge工具方法来实现多个流的组合，这个也是通用的：
//多个AsyncSequence merge成一个AsyncSequence
func mergeSequence<Seq : AsyncSequence>(seqs : [Seq]) -> AsyncThrowingStream<Seq.Element, Error>
makeSequence(f: mergeF(fs: seqs.map(makeLoadFunc)))
func makeLoadFunc<Seq : AsyncSequence>(ats : Seq) -> StreamFunc<Seq.Element>
onData, onError, onComplete in
Task
do
for try await data in ats
onData(data)
onComplete()
catch
onError(error)
func mergeF<Data>(fs : [StreamFunc<Data>]) -> StreamFunc<Data>
onData, onError, onComplete in
var finish = false
var results = fs.map_ in false
for (index, f) in fs.enumerated()
f data in
if (!finish)
onData(data)
_: e in
// 如果是第一个错误，直接回调，同时标记为结束
if (!finish)
finish = true
onError(e)
_:
// 注意，即使所有的请求都结束了，回调成功
if (!finish)
results[index] = true
// 所有都结束，回调
if (!results.contains(false))
finish = true
onComplete()
第三步，将listOrder和queryUserInfo转换成StreamFunc形式，与Rx中的第二步实现完全相同；
第四步，这样就可以将load方法简化为：
func asLoad() -> AsyncThrowingStream<[OrderObject], Error>
let orderService = [OrderService("鹅鹅鹅"), OrderService("鸭鸭鸭")]
// 通过map构造AsyncSequence，通过flatMap对listOrder和queryUserInfo进行复合
let streams = orderService.map orderService in
makeSequence(f: orderService.listOrders).flatMap (orders) -> AsyncThrowingStream<[OrderObject], Error> in
makeSequence(f: makeLoadFunc(orders: orders, userInfoService: getUserService(site: orderService.site)))
// merge两个平台的AsyncSequence
return mergeSequence(seqs: streams)
可以发现，代码与RxSwift几乎是完全相同的，所以我们仍然有对于代码正确性的信心，不同的是，现在使用方也得以获得同样的信心：
for try await orderObject in asLoad()
print("async get orderObject \\(orderObject.first?.order.orderId)")
总结
同步是编程中的田园世界，而流式接口作为异步接口最复杂的形态，我们通过CPS变换的控制流技术，将流式接口表示为AsyncSequence，实现了对异步序列遍历的同步形式，从而将响应式编程在形式上统一回了田园世界。
上面的第一步和第二步实现了AsyncSequence和StreamFunc的相互转换，所以实际上我们证明了它们是同构的，更进一步的，我们可以证明它们与Rx、Combine也是同构的。换言之，它们是同一个概念的不同形式，理论上它们的表达能力是等价的，这个概念就是数据流，这个概念在Rx中叫做Observable，在Combine中叫做Publisher。
在实际实现上，Rx和Combine提供了大量的操作符，因此目前它们的能力远远强于AsyncSequence和StreamFunc，比如AsyncSequence居然不支持merge。
AsyncSequence的优势是可以支持同步写法，在我看来这个优势是很大的。看到社区有过AsyncSequence替换Combine的相关的讨论，我认为逻辑上是讲得通的。
AsyncSequence替换Combine的相关讨论地址：https://forums.swift.org/t/should-asyncsequence-replace-combine-in-the-future-or-should-they-coexist/53370）
团队介绍
我们是来自淘宝全域触达&用户互动客户端团队，负责包含Push、POP弹层和消息沟通三大触达场景。全域触达&用户互动客户端团队追求极致的性能、流畅的交互体验和稳定的触达效率，用智能化的调控策略为用户带来更好的使用体验。
✿ 拓展阅读
作者|蒋文豪（四点）
编辑|橙子君
以上是关于响应式编程的复杂度和简化的主要内容，如果未能解决你的问题，请参考以下文章