摘要：的讀寫流模塊提供來創建行讀取流，即讀取文件的每一行作為持續的輸入數據模塊提供來創建寫入流，其返回的有和兩個方法，來完成流式的寫入與結束寫入。

前幾天接到任務要使用第三方API處理幾千張圖片，得到結果集。我的做法就是使用Rx.js結合node的讀寫流來完成數據讀入、接口請求、數據處理、數據寫入這些操作。本篇就來分享這個代碼和其邏輯。

Rx.js是一個響應式編程庫，能簡化事件／異步處理邏輯代碼。其視所有的事件／數據為__流__，提供各種流處理的operators，將輸入與輸出平滑的鏈接起來。可以類比為linux上的pipe操作符: ls | grep a*b | less。

readline模塊提供readline.createInterface來創建行讀取流，即讀取文件的每一行作為持續的輸入數據

fs模塊提供fs.createWriteStream來創建寫入流， 其返回的writer有write和end兩個方法，來完成流式的寫入與結束寫入。

并發數有限制，3個是出現其出現并發錯誤概率最低的最大并發數

接口請求過于頻繁，會較大概率出現連續的并發錯誤， 大概延遲400秒效果尚可

提供給第三方的圖片是鏈接，其需要服務器自己下載，會出現操作超時或者長時間不返回的情況。

從文件讀取圖片文件名

拼接url

發送3個并發請求

請求出現超時問題重試3次，最后如果失敗則放棄

出現非超時錯誤（如并發錯誤等）則一直重試，直到成功

請求成功后延遲400秒繼續發起下一個請求

處理返回的數據

寫入文件

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

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

const writeStream = fs.createWriteStream("output.txt");  // 創建寫入流

Rx.Observable.fromEvent(imgStream, "line")  // 將行讀取流轉化為Rx的事件流
.takeUntil(Rx.Observable.fromEvent(imgStream, "close"))  // 讀取流截止時終止Rx流
.map(img => generateData(img))  // 將文件名處理成post的數據
 // 發起請求，并發3個，請求返回后延遲400ms后再進行下一步處理并發起下一個請求
.mergeMap(data => requestAPI(data).delay(400), (o, i) => i, 3) 
.subscribe(data => {
    // 處理數據并寫入文件
    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();
});

其中的需要關注的點在.mergeMap(data => requestAPI(data).delay(400), (o, i) => i, 3) ，這里內部requestAPI返回一個封裝了http異步請求并延遲400ms的Rx流，當請求完成并延遲完成后將數據返回上一層繼續進行處理（可以類比為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();  // 觸發重試的流，當其發出數據時會使`retryWhen`觸發重試錯誤流
    let retryTimes = 3;  // 設置非正常失敗（超時）重試的上限

    // 使用Rx的自定義流封裝一個帶錯誤重試機制的http請求，可以類比為new Promise
    // 但要注意的是Rx是流，即數據是可以持續的，而Promise則只有一個結果和狀態
    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) {
                    // 請求正常返回，向流內推送結果并結束
                    observer.next({
                        status: res.statusCode,
                        url: reqData.image_url,
                        data: JSON.parse(data)
                    });
                    observer.complete();
                } else {
                    // 請求正常返回，但不是正常結果，拋出錯誤并重試
                    console.log(`retring: ${reqData.image_url}`);
                    observer.error({
                        status: res.statusCode,
                        url: reqData.image_url
                    });
                    retry$.next(true);
                }
            });
        });

        req.setTimeout(4000, () => {
            // 設置請求4s超時，超時后終止，引發請求拋錯
            req.abort();
        });

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

        req.write(dataStr);

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

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

本文始發于本人的公眾號：楓之葉。公眾號二維碼