摘要：的讀寫(xiě)流模塊提供來(lái)創(chuàng)建行讀取流，即讀取文件的每一行作為持續(xù)的輸入數(shù)據(jù)模塊提供來(lái)創(chuàng)建寫(xiě)入流，其返回的有和兩個(gè)方法，來(lái)完成流式的寫(xiě)入與結(jié)束寫(xiě)入。

前幾天接到任務(wù)要使用第三方API處理幾千張圖片，得到結(jié)果集。我的做法就是使用Rx.js結(jié)合node的讀寫(xiě)流來(lái)完成數(shù)據(jù)讀入、接口請(qǐng)求、數(shù)據(jù)處理、數(shù)據(jù)寫(xiě)入這些操作。本篇就來(lái)分享這個(gè)代碼和其邏輯。

Rx.js是一個(gè)響應(yīng)式編程庫(kù)，能簡(jiǎn)化事件／異步處理邏輯代碼。其視所有的事件／數(shù)據(jù)為_(kāi)_流__，提供各種流處理的operators，將輸入與輸出平滑的鏈接起來(lái)?？梢灶?lèi)比為linux上的pipe操作符: ls | grep a*b | less。

readline模塊提供readline.createInterface來(lái)創(chuàng)建行讀取流，即讀取文件的每一行作為持續(xù)的輸入數(shù)據(jù)

fs模塊提供fs.createWriteStream來(lái)創(chuàng)建寫(xiě)入流， 其返回的writer有write和end兩個(gè)方法，來(lái)完成流式的寫(xiě)入與結(jié)束寫(xiě)入。

并發(fā)數(shù)有限制，3個(gè)是出現(xiàn)其出現(xiàn)并發(fā)錯(cuò)誤概率最低的最大并發(fā)數(shù)

接口請(qǐng)求過(guò)于頻繁，會(huì)較大概率出現(xiàn)連續(xù)的并發(fā)錯(cuò)誤， 大概延遲400秒效果尚可

提供給第三方的圖片是鏈接，其需要服務(wù)器自己下載，會(huì)出現(xiàn)操作超時(shí)或者長(zhǎng)時(shí)間不返回的情況。

從文件讀取圖片文件名

拼接url

發(fā)送3個(gè)并發(fā)請(qǐng)求

請(qǐng)求出現(xiàn)超時(shí)問(wèn)題重試3次，最后如果失敗則放棄

出現(xiàn)非超時(shí)錯(cuò)誤（如并發(fā)錯(cuò)誤等）則一直重試，直到成功

請(qǐng)求成功后延遲400秒繼續(xù)發(fā)起下一個(gè)請(qǐng)求

處理返回的數(shù)據(jù)

寫(xiě)入文件

const https = require("https");
const querystring = require("querystring");
const Rx = require("rxjs");
const readline = require("readline");
const fs = require("fs");

const imgStream = readline.createInterface({  // 創(chuàng)建行讀取流
    input: fs.createReadStream("filelist.txt")
});

const writeStream = fs.createWriteStream("output.txt");  // 創(chuàng)建寫(xiě)入流

Rx.Observable.fromEvent(imgStream, "line")  // 將行讀取流轉(zhuǎn)化為Rx的事件流
.takeUntil(Rx.Observable.fromEvent(imgStream, "close"))  // 讀取流截止時(shí)終止Rx流
.map(img => generateData(img))  // 將文件名處理成post的數(shù)據(jù)
 // 發(fā)起請(qǐng)求，并發(fā)3個(gè)，請(qǐng)求返回后延遲400ms后再進(jìn)行下一步處理并發(fā)起下一個(gè)請(qǐng)求
.mergeMap(data => requestAPI(data).delay(400), (o, i) => i, 3) 
.subscribe(data => {
    // 處理數(shù)據(jù)并寫(xiě)入文件
    let str = data.url;
    if (data.status === 200 && data.data.xxx.length) {
        zzz = data.data.xxx.map(x => x.zzz);
        str += `    ${JSON.stringify(zzz)}`;
    }
    writeStream.write(`${str}
`);
}, (err) => {
    console.log(err);
    console.log("!!!!!!!!!!!ERROR!!!!!!!!!");
}, () => {
    console.log("=====complete======");
    writeStream.end();
});

其中的需要關(guān)注的點(diǎn)在.mergeMap(data => requestAPI(data).delay(400), (o, i) => i, 3) ，這里內(nèi)部requestAPI返回一個(gè)封裝了http異步請(qǐng)求并延遲400ms的Rx流，當(dāng)請(qǐng)求完成并延遲完成后將數(shù)據(jù)返回上一層繼續(xù)進(jìn)行處理（可以類(lèi)比為Promise的then)

const requestFacepp = dataStr => {
    const options = {
        hostname: "api.xxx.com",
        port: 443,
        path: "/xxx",
        method: "POST",
        headers: {
            "Content-Type": "application/x-www-form-urlencoded",
            "Content-Length": Buffer.byteLength(dataStr)
        }
    };
    const reqData = querystring.parse(dataStr);
    const retry$ = new Rx.Subject();  // 觸發(fā)重試的流，當(dāng)其發(fā)出數(shù)據(jù)時(shí)會(huì)使`retryWhen`觸發(fā)重試錯(cuò)誤流
    let retryTimes = 3;  // 設(shè)置非正常失敗（超時(shí)）重試的上限

    // 使用Rx的自定義流封裝一個(gè)帶錯(cuò)誤重試機(jī)制的http請(qǐng)求，可以類(lèi)比為new Promise
    // 但要注意的是Rx是流，即數(shù)據(jù)是可以持續(xù)的，而Promise則只有一個(gè)結(jié)果和狀態(tài)
    return Rx.Observable.create(observer => {
        const req = https.request(options, res => {
            let data = "";
            res.setEncoding("utf8");
            res.on("data", chunk => {
                data += chunk;
            });
            res.on("end", () => {
                if (res.statusCode === 200) {
                    // 請(qǐng)求正常返回，向流內(nèi)推送結(jié)果并結(jié)束
                    observer.next({
                        status: res.statusCode,
                        url: reqData.image_url,
                        data: JSON.parse(data)
                    });
                    observer.complete();
                } else {
                    // 請(qǐng)求正常返回，但不是正常結(jié)果，拋出錯(cuò)誤并重試
                    console.log(`retring: ${reqData.image_url}`);
                    observer.error({
                        status: res.statusCode,
                        url: reqData.image_url
                    });
                    retry$.next(true);
                }
            });
        });

        req.setTimeout(4000, () => {
            // 設(shè)置請(qǐng)求4s超時(shí)，超時(shí)后終止，引發(fā)請(qǐng)求拋錯(cuò)
            req.abort();
        });

        req.on("error", err => {
            console.log(`retring(${retryTimes}): ${reqData.image_url}`);
            // 請(qǐng)求拋錯(cuò)時(shí)重試，超出次數(shù)則終止本次請(qǐng)求
            observer.error(`error: ${err.message}`);
            if (retryTimes > 0) {
                retryTimes--;
                retry$.next(true);
            } else {
                retry$.complete();
            }
        });

        req.write(dataStr);

        req.end();
        return () => { req.abort() };  // 返回終止流的處理回調(diào)
    })
    .retryWhen(errs => errs.switchMap(err => {
        // 未超過(guò)次數(shù)返回重試流，超出則返回錯(cuò)誤數(shù)據(jù)并終止本次Rx流
        return retryTimes > 0 ? retry$ : Rx.Observable.of({
            status: 500,
            url: reqData.image_url
        });
    }));
};

到此就搬磚完畢，開(kāi)個(gè)車(chē)讓他慢慢跑就可以了。
本篇展示了Rx在流數(shù)據(jù)處理與異步處理上的方式，邏輯與代碼都挺清晰、扁平。在處理交雜的邏輯時(shí)也不錯(cuò)（重試部分）。如果喜歡或者有幫助的話(huà)可以后面在發(fā)一篇Rx在復(fù)雜DOM事件處理上的應(yīng)用。;-)

本文始發(fā)于本人的公眾號(hào)：楓之葉。公眾號(hào)二維碼