这一篇是在实际工程中遇到的一个难得的例子；反映在Node里两种编程范式的设计冲突。这种冲突具有普适性，但本文仅分析问题本质，不探讨更高层次的抽象。

我在写一个类似HTTP的资源协议，叫RP，Resource Protocol，和HTTP不同的地方，RP是构建在一个中立的传输层上的；这个传输层里最小的数据单元，message，是一个JSON对象。

协议内置支持multiplexing，即一个传输层连接可以同时维护多个RP请求应答过程。

考虑客户端request类设计，类似Node内置的HTTP Client，或流行的npm包，如request或 superagent；

可以采用EventEmitter方式emit error和responses事件，也可以采用Node Callback的形式，需要使用者提供接口形式为(err, res) => {}的callback函数。

随着async/await的流行，request类也可以提供一个.then接口，用如下方式实现（实际上superagent就是这么实现的）：

class Request extends Duplex {
    constructor () {
        super()
        ...
        this.promise = new Promise((resolve, reject) => {
            this.resolve = resolve
            this.reject = reject
        })
    }
    
    then (...args) {
        return this.promise.then(...args)
    }
}

RP的实际设计，形式和大家熟悉的HTTP Client有一点小区别，response对象本身不是stream，而是把stream做为一个property提供。换句话说，callback函数形式为：

(err, { data, chunk, stream }) => {}

如果请求返回的不是stream，则data或者chunk有值；如果返回的是stream，则仅stream有值，且为stream.Readable类型。

这个形式上的区别和本文要讨论的问题无关。

RP底层从传输层取二进制数据，解析出message，然后emit给上层；它采用了一个简单方式，循环解析收到的data chunk，直到没有完整的message为止。

这意味着可以在一个tick里分发多个消息。request对象也必须能够在一个tick里处理多个来自服务端的消息。

我们具体要讨论的情况是服务器连续发了这样两条消息：

1. status 200 with stream
2. abort

第一条意思是后面还有message stream，第二条abort指server发生意外无法继续发送了。

在request对象收到第一条消息时，它创建response对象，包含stream对象：

this.res = { stream: new stream.Readabe({...}) }
// this.emit('response', this.res)
// this.callback(null, this.res)
this.resolve(this.res)

象注释中emit或trigger使用者提供的callback，都没有问题；但如果调用resolve，注意，Promise是保证异步的，这意味着使用者通过then提供的onFulfilled，不会在当前tick被调用。

接下来第二条消息，abort，在同一个tick被处理；但这个时候，因为使用者还没来得及挂载上任何listener，包括error handler，如果设计上要求这个stream emit error——很合理的设计要求——此时，按照Node的约定，error没有handler，整个程序crash了。

这个问题的dirty fix有很多种办法。

首先request.handleMessage方法，如果无法同步完成对message的处理，而message的处理顺序又需要保证，它应该buffer message，这是node里最常见的一种synchronize方式，代表性的实现就是stream.Writable。

但这里有一个困难，this.resolve这个函数没有callback提供，必须预先知道运行环境的Promise实现方式；在node里是nextTick，所以在this.resolve之后nextTick一下，同时buffer其它后续消息的处理，可以让使用者在onFulfilled函数中给stream挂载上handler。

这里可以看出，callback和emitter实际上是同步的。

当调用callback或者listener时，request和使用者做了一个约定，你必须在这个函数内做什么（在对象上挂载所有的listener），然后我继续做什么（处理下一个消息，emit data或者error）；这相当于是interface protocol对顺序的约定。

我们可以称之为synchronous sequential composition，是程序语义意义上的。

对应的asynchronous版本呢？

如果我们不去假设运行环境的Promise的实现呢？它应该和同步版本的语义一样对吧。

再回头看看问题，假如stream emit error不会导致系统crash，使用者在onFulfilled拿到{ stream }这个对象时，它看到了什么？一个已经发生错误后结束了的stream。

这个可能使用上会难过一点，需要判断一下，但还感觉不出是多大的问题。

再进一步，如果是另一种情况呢？Server在一个chunk里发来了3个消息；

1. status 200 with stream
2. data
3. abort

这个时候使用者看到的还是一个errored stream，data去哪里了呢？你还能说asynchronous sequential composition的语义和synchronous的一致么？不能了对吧，同步的版本处理了data，很可能对结果产生影响。

在理想的情况下，sequential composition，无论是synchronous的，还是asynchronous的，语义（执行结果）应该一致。

那么来看看如何做到一个与Promise A+的实现无关的做法，保证异步和同步行为一致。

如果你愿意用『通讯』理解计算，这个问题的答案很容易思考出来：假想这个异步的handler位于半人马座阿尔法星上，那我们唯一能做的事情是老老实实按照事件发生的顺序，发送给它，不能打乱顺序，就像我们收到他们时一样。

但是当我们把进来的message，翻译实现成stream时，没能保证这个order，包括：

1. abort消息抢先/乱序
2. data消息丢失了

这是问题的root cause，当我们异步处理一个消息序列时，前面写的实现break了顺序和内容的完整性。

在数学思维上，我们说Promise增加了一个callback/EventEmitter不具备的属性，deferred evaluation，是一个编程中罕见的temporal属性；当然这不奇怪，因为这就是Promise的目的。

同时Promise -> Value还有一个属性是它可以被不同的使用者访问多次，保持了Value的属性。

这也不奇怪。

只是Stream作为一种体积上可以为无穷大的值，在实践中不可能去cache所有的值，把它整体当成一个值处理，所以这个可以被无限提取的『值』属性就消失了。

但是这不意味着stream作为一个对象，它的行为，不能延迟等到它被构造且使用后才开始处理消息。

一种方式是写一个stream有这种能力的；stream.Readable有一个flow属性，必须通过readable.resume开始，这是一个触发方式；另一个方式是有点tricky，可以截获response.stream的getter，在它第一次被访问时触发异步处理buffered message。

这样的做法是不需要依赖Promise A+的实现的；但不是百分百asynchronous sequential composition，因为stream的handler肯定是synchronous的。

完全的asynchronous可以参照Dart的使用await消费stream的方式。

它的逻辑可以这样理解：把所有Event，无论哪里来的，包括error，都写到一个流里去，用await消费这个流；但实际上在await返回的时候仍然面对一个状态机，好处是

1. throw给力；
2. 流程等待方便，即处理流输出的对象时还可以有await语句，在取下一个流输出的对象之前，相当于一种blocking；但这种blocking需要慎重，它是反并发的；

总结：

Node的Callback和EventEmitter在组合时handler/listener是同步的；Promise则反过来保证每个handler/listener都是异步组合，这是两者的根本区别。

在顺序组合函数（或者进程代数意义上的进程）上，同步组合是紧耦合的；它体现在一旦功能上出现什么原因，需要把一个同步逻辑修改成异步时，都要大动干戈，比如本来是读取内存，后来变成了读取文件。

如果程序天生写成异步组合，类似变化就不会对实现逻辑产生很大影响；但是细粒度的异步组合有巨大的性能损失，这和现代处理器和编译器的设计与实现有关。

真正理想的情况应该是开发者只表达“顺序”，并不表达它是同步还是异步实现；就像前面看到的，实际上同步的实现都有可以对应的异步实现，差别只是执行效率和内存使用（buffer有更多的内存开销，同步处理实际上更多是『阅后即焚』）；

但我们使用的imperative langugage不是如此，它在强制你表达顺序；而另外一类号称未来其实狗屎的语言，在反过来强制你不得表达顺序。

都是神经病。学术界就不会真正理解产业界的实际问题。

原文：https://segmentfault.com/a/1190000022609354

链接: https://fly63.com/article/detial/8950