摘要：場(chǎng)景實(shí)際業(yè)務(wù)中可能出現(xiàn)重復(fù)消費(fèi)一個(gè)可讀流的情況，比如在前置過(guò)濾器解析請(qǐng)求體，拿到進(jìn)行相關(guān)權(quán)限及身份認(rèn)證認(rèn)證通過(guò)后框架或者后置過(guò)濾器再次解析請(qǐng)求體傳遞給業(yè)務(wù)上下文。

實(shí)際業(yè)務(wù)中可能出現(xiàn)重復(fù)消費(fèi)一個(gè)可讀流的情況，比如在前置過(guò)濾器解析請(qǐng)求體，拿到body進(jìn)行相關(guān)權(quán)限及身份認(rèn)證；認(rèn)證通過(guò)后框架或者后置過(guò)濾器再次解析請(qǐng)求體傳遞給業(yè)務(wù)上下文。因此，重復(fù)消費(fèi)同一個(gè)流的需求并不奇葩，這類似于js上下文中通過(guò) deep clone一個(gè)對(duì)象來(lái)操作這個(gè)對(duì)象副本，防止源數(shù)據(jù)被污染。

const Koa = require("koa");
const app = new Koa();

let parse = function(ctx){
    return new Promise((res)=>{
        let chunks = [],len  = 0, body = null;
        ctx.req.on("data",(chunk)=>{
            chunks.push(chunk)
            len += chunk.length
        });
        ctx.req.on("end",()=>{
            body = (Buffer.concat(chunks,len)).toString();
            res(body);
        });
    })
}
// 認(rèn)證
app.use(async (ctx,next) => {
    let body = JSON.parse(decodeURIComponent(await parse(ctx)));
    if(body.name != "admin"){
        return ctx.body = "permission denied!"
    }
    await next();
})
// 解析body體，傳遞給業(yè)務(wù)層
app.use(async (ctx,next) => {
    let body = await parse(ctx);
    ctx.postBody = body;
    await next();
})
app.use(async ctx => {
  ctx.body = "Hello World
";
  ctx.body += `post body: ${ctx.postBody}`;
});

app.listen(3000);

上述代碼片段無(wú)法正常運(yùn)行，請(qǐng)求無(wú)法得到響應(yīng)。這是因?yàn)樵谇爸眠^(guò)濾器的認(rèn)證邏輯中消費(fèi)了請(qǐng)求體，在第二級(jí)過(guò)濾器中就無(wú)法再次消費(fèi)請(qǐng)求體，因此請(qǐng)求會(huì)阻塞。實(shí)際業(yè)務(wù)中，認(rèn)證邏輯往往是與每個(gè)公司規(guī)范相關(guān)的，是一個(gè)“二方庫(kù)”；而示例中的第二季過(guò)濾器則通常作為一個(gè)三方庫(kù)存在，因此為了不影響第三方包消費(fèi)請(qǐng)求體，必須在認(rèn)證的二方包中保存 ctx.req 這個(gè)可讀流的數(shù)據(jù)仍然存在，這就涉及到本文的主旨了。

復(fù)制流并不像復(fù)制一個(gè)對(duì)象一樣簡(jiǎn)單與直接，流的使用是一次性的，一旦一個(gè)可讀流被消費(fèi)（寫(xiě)入一個(gè)Writeable對(duì)象中），那么這個(gè)可讀流就是不可再生的，無(wú)法再使用?？墒峭ㄟ^(guò)一些簡(jiǎn)單的技巧可以再次復(fù)原一個(gè)可讀流，不過(guò)這個(gè)復(fù)原出來(lái)的流雖然內(nèi)容和之前的流相同，但卻不是同一個(gè)對(duì)象了，因此這兩個(gè)對(duì)象的屬性及原型都不同，這往往會(huì)影響后續(xù)的使用，不過(guò)辦法總是有的，且看下文。

可讀流的“影分身之術(shù)”和鳴人的差不多，不過(guò)僅限于被克隆對(duì)象的 流 這一特性，即保證克隆出的流有著相同的數(shù)據(jù)。但是克隆出來(lái)的流卻無(wú)法擁有原對(duì)象的其他屬性，但我們可通過(guò)原型鏈繼承的方式實(shí)現(xiàn)屬性及方法的繼承。

let Readable = require("stream").Readable;
let fs = require("fs");
let path = require("path");

class NewReadable extends Readable{
    constructor(originReadable){
        super();
        this.originReadable = originReadable;
        this.start();
    }

    start() {
        this.originReadable.on("data",(chunck)=>{
            this.push(chunck);
        });

        this.originReadable.on("end",()=>{
            this.push(null);
        });
        
        this.originReadable.on("error",(e)=>{
            this.push(e);
        });
    }

    // 作為Readable的實(shí)現(xiàn)類，必須實(shí)現(xiàn)_read函數(shù)，否則會(huì)throw Error
    _read(){
    }
}

app.use(async (ctx,next) => {
    let cloneReq = new NewReadable(ctx.req);
    let cloneReq2 = new NewReadable(ctx.req);
    // 此時(shí)，ctx.req已被消費(fèi)完(沒(méi)有內(nèi)容)，所有的數(shù)據(jù)都完全在克隆出的兩個(gè)流上

    // 消費(fèi)cloneReq，獲取認(rèn)證數(shù)據(jù)
    let body = JSON.parse(decodeURIComponent(await parse({req: cloneReq})));

    // 將克隆出的cloneReq2重新設(shè)置原型鏈，繼承ctx.req原有屬性
    cloneReq2.__proto__ = ctx.req;
    // 此后重新給ctx.req復(fù)制，留給后續(xù)過(guò)濾器消費(fèi)
    ctx.req = cloneReq2;

    if(body.name != "admin"){
        return ctx.body = "permission denied!"
    }
    await next();
})

點(diǎn)評(píng): 這種影分身之術(shù)可以同時(shí)復(fù)制出多個(gè)可讀流，同時(shí)需要針對(duì)原來(lái)的流重新進(jìn)行賦值，并繼承原有屬性，這樣才能不影響后續(xù)的重復(fù)消費(fèi)。

stream模塊有一個(gè)特殊的類，即 Transform。關(guān)于Transfrom的特性，我曾在 深入node之Transform 一文中詳細(xì)介紹過(guò)，他擁有可讀可寫(xiě)流雙重特性，那么利用Transfrom可以快速簡(jiǎn)單的實(shí)現(xiàn)克隆。

首先，通過(guò) pipe 函數(shù)將可讀流導(dǎo)向兩個(gè) Transform流（之所以是兩個(gè)，是因?yàn)樾枰谇爸眠^(guò)濾器消費(fèi)一個(gè)流，后續(xù)的過(guò)濾器消費(fèi)第二個(gè)）。

let cloneReq = new Transform({
    highWaterMark: 10*1024*1024,
    transform: (chunk,encode,next)=>{
        next(null,chunk);
    }
});
let cloneReq2 = new Transform({
    highWaterMark: 10*1024*1024,
    transform: (chunk,encode,next)=>{
        next(null,chunk);
    }
});
ctx.req.pipe(cloneReq)
ctx.req.pipe(cloneReq2)

上述代碼中，看似 ctx.req 流被消費(fèi)（pipe）了兩次，實(shí)際上 pipe 函數(shù)則可以看成 Readable和Writeable實(shí)現(xiàn)backpressure的一種“語(yǔ)法糖”實(shí)現(xiàn)，具體可通過(guò) node中的Stream－Readable和Writeable解讀 了解，因此得到的結(jié)果就是“ctx.req被消費(fèi)了一次，可是數(shù)據(jù)卻復(fù)制在cloneReq和cloneReq2這兩個(gè)Transfrom對(duì)象的讀緩沖區(qū)里，實(shí)現(xiàn)了clone”

其實(shí)pipe針對(duì)Readable和Writeable做了限流，首先針對(duì)Readable的data事件進(jìn)行偵聽(tīng)，并執(zhí)行Writeable的write函數(shù)，當(dāng)Writeable的寫(xiě)緩沖區(qū)大于一個(gè)臨界值（highWaterMark），導(dǎo)致write函數(shù)返回false（此時(shí)意味著Writeable無(wú)法匹配Readable的速度，Writeable的寫(xiě)緩沖區(qū)已經(jīng)滿了），此時(shí)，pipe修改了Readable模式，執(zhí)行pause方法，進(jìn)入paused模式，停止讀取讀緩沖區(qū)。而同時(shí)Writeable開(kāi)始刷新寫(xiě)緩沖區(qū)，刷新完畢后異步觸發(fā)drain事件，在該事件處理函數(shù)中，設(shè)置Readable為flowing狀態(tài)，并繼續(xù)執(zhí)行flow函數(shù)不停的刷新讀緩沖區(qū)，這樣就完成了pipe限流。需要注意的是，Readable和Writeable各自維護(hù)了一個(gè)緩沖區(qū)，在實(shí)現(xiàn)的上有區(qū)別：Readable的緩沖區(qū)是一個(gè)數(shù)組，存放Buffer、String和Object類型；而Writeable則是一個(gè)有向鏈表，依次存放需要寫(xiě)入的數(shù)據(jù)。

最后，在數(shù)據(jù)復(fù)制的同時(shí)，再給其中一個(gè)對(duì)象復(fù)制額外的屬性即可：

// 將克隆出的cloneReq2重新設(shè)置原型鏈，繼承ctx.req原有屬性
cloneReq2.__proto__ = ctx.req;
// 此后重新給ctx.req復(fù)制，留給后續(xù)過(guò)濾器消費(fèi)
ctx.req = cloneReq2;

至此，通過(guò)Transform實(shí)現(xiàn)clone已完成。完整的代碼如下(最前置過(guò)濾器)：

// 認(rèn)證
app.use(async (ctx,next) => {
    // let cloneReq = new NewReadable(ctx.req);
    // let cloneReq2 = new NewReadable(ctx.req);
    let cloneReq = new Transform({
        highWaterMark: 10*1024*1024,
        transform: (chunk,encode,next)=>{
            next(null,chunk);
        }
    });
    let cloneReq2 = new Transform({
        highWaterMark: 10*1024*1024,
        transform: (chunk,encode,next)=>{
            next(null,chunk);
        }
    });
    ctx.req.pipe(cloneReq)
    ctx.req.pipe(cloneReq2)
    // 此時(shí)，ctx.req已被消費(fèi)完(沒(méi)有內(nèi)容)，所有的數(shù)據(jù)都完全在克隆出的兩個(gè)流上

    // 消費(fèi)cloneReq，獲取認(rèn)證數(shù)據(jù)
    let body = JSON.parse(decodeURIComponent(await parse({req: cloneReq})));

    // 將克隆出的cloneReq2重新設(shè)置原型鏈，繼承ctx.req原有屬性
    cloneReq2.__proto__ = ctx.req;
    // 此后重新給ctx.req復(fù)制，留給后續(xù)過(guò)濾器消費(fèi)
    ctx.req = cloneReq2;

    if(body.name != "admin"){
        return ctx.body = "permission denied!"
    }
    await next();
})

說(shuō)明

ctx.req執(zhí)行兩次pipe到對(duì)應(yīng)cloneReq和cloneReq2，然后立即消費(fèi)cloneReq對(duì)象，這樣合理嗎？如果源數(shù)據(jù)夠大，pipe還未結(jié)束就在消費(fèi)cloneReq，會(huì)不會(huì)有什么問(wèn)題？

其實(shí) pipe函數(shù)里面大多是異步操作，即針對(duì) 源和目的流做的一些流控措施。目的流使用的是cloneReq對(duì)象，該對(duì)象在實(shí)例化的過(guò)程中 transform函數(shù)直接通過(guò)調(diào)用next函數(shù)將接受到的數(shù)據(jù)傳入到Transform對(duì)象的可讀流緩存中，同時(shí)觸發(fā)‘readable和data事件’。這樣，我們?cè)谙挛南M(fèi)cloneReq對(duì)象也是通過(guò)“偵聽(tīng)data事件”實(shí)現(xiàn)的，因此即使ctx.req的數(shù)據(jù)仍沒(méi)有被消費(fèi)完，下文仍可以正常消費(fèi)cloneReq對(duì)象。數(shù)據(jù)流仍然可以看做是從ctx.req --> cloneReq --> 消費(fèi)。

使用Transform流實(shí)現(xiàn)clone 可讀流的弊端：

上例中，Transfrom流的實(shí)例化傳入了一個(gè)參數(shù) highWaterMark,該參數(shù)在Transfrom中的作用 在 上文 深入node之Transform 中有過(guò)詳解，即當(dāng)Transfrom流的讀緩沖大小 < highWaterMark時(shí)，Transfrom流就會(huì)將接收到的數(shù)據(jù)存儲(chǔ)在讀緩沖里，等待消費(fèi)，同時(shí)執(zhí)行 transfrom函數(shù)；否則什么都不做。

因此，當(dāng)要clone的源內(nèi)容大于highWaterMark時(shí)，就無(wú)法正常使用這種方式進(jìn)行clone了，因?yàn)橛捎谠磧?nèi)容>highWaterMark，在沒(méi)有后續(xù)消費(fèi)Transfrom流的情況下就不執(zhí)行transfrom方法（當(dāng)Transfrom流被消費(fèi)時(shí)，Transfrom流的讀緩沖就會(huì)變小，當(dāng)其大小

所以設(shè)置一個(gè)合理的highWaterMark大小很重要，默認(rèn)的highWaterMark為 16kB。