响应式编程的复杂度和简化

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点本身的复杂度和它引起的众多问题已经非常熟悉了。



这个例子很简单,只有三步:

  1. 假设需要为一个店铺提供一个订单展示页面,这些订单来自两个不同的平台“鹅鹅鹅”和“鸭鸭鸭”,他们各自提供了查询的接口(listOrders,为了简单假设他们提供的模型和接口完全一致);  

  2. 订单列表需要展示用户的昵称等信息,需要通过对应平台的另外一个接口(queryUserInfo)查询;  

  3. 由于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的请求中也做类似的处理,并保证错误回调后不再执行任何回调。



完成的回调要比错误复杂的多,我们可以来思考一下:

  1. 首先,我们不能在listOrders的onComplete里面取回调完成,因为这里不能代表queryUserInfo这个接口也完成了;  

  2. 其次,我们也不能简单的通过所有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弹层和消息沟通三大触达场景。全域触达&用户互动客户端团队追求极致的性能、流畅的交互体验和稳定的触达效率,用智能化的调控策略为用户带来更好的使用体验。

✿  拓展阅读

作者|蒋文豪(四点)

编辑|橙子君

以上是关于响应式编程的复杂度和简化的主要内容,如果未能解决你的问题,请参考以下文章

Jetpack Compose开源:基于Kotlin的响应式编程方案,简化UI开发

IOS函数响应式编程开发简析

函数响应式编程的“信号”表示是不是正确?

RxSwift 响应式编程详解(001)

Flutter 应用开发之Bloc模式

学习推荐5:Spring响应式编程实战