大數據採集中的異步處理問題

時間 2019-11-10

原文原文鏈接

這段時間在學習nodejs，用jsdom採集了一些數據，也遇到了一些問題，貼出來徵求一下你們的解決方案。node

首先說一下目的，有幾十萬條圖片數據，把這些圖片抓取到本地文件夾中，採集完成後把成功數據歸檔爲done.json,失敗數據歸檔爲undone.json，以下：json

1 const IMGS = [
2     "http://xxx.com/1.jpg",
3     "http://xxx.com/2.jpg",
4     "http://xxx.com/3.jpg",
5     "http://xxx.com/4.jpg",
6     ...//此處省略無數的數據
7 ];

採集的心路歷程：promise

方法1 forEach循環直接請求（失敗）網絡

1 IMGS.forEach(v=>{
2      http.get(v, (res)=> {
3          //blablabla
4      });
5 });

這種方法一會兒就把全部數據都異步的請求出去了，超出了網絡連接數，直接就把服務爆掉了。併發

方法2 採用遞歸處理完一條自動處理下一條。(成功，可是有問題）dom

 1 const DONE = [],
 2       UNDONE = [],
 3       FILEDONE = './done.json',
 4       FILEUNDONE = './undone.json';
 5 let counter = 0;
 6 const writeResult() =>{
 7     fs.writeFile(FILEDONE, JSON.stringify(DONE), "utf8", ()=>console.log('done saved!'));
 8     fs.writeFile(FILEUNDONE , JSON.stringify(UNDONE), "utf8", ()=>console.log('undone saved!'));
 9 }
10 const saveImage = (data, fn)=> {
11     fs.writeFile(counter++ + '.jpg', data, fn);
12 }
13 const downloadImage = (i)=> {
14     if(i >= IMGS.length) {
15         writeResult();
16         return;
17     }
18     http.get(IMGS[i], (res)=> {
19         //此處略去了異常處理的內容
20         res.setEncoding="binary";
21         let dt = '';
22         res.on('data', (ck)=> dt+=ck;)
23         res.on('end',()=> {
24             saveImage(dt, ()=> {
25                 DONE.push(IMGS[i]);
26                 downloadImag(i++);
27             });
28         })
29         
30     }).on('error',()=>UNDONE.push(IMGS[i])) ;
31 }
32 
33 downloadImage(0);

這種方法的確是能成功抓取到了數據，可是有一個最大的弊端就是隻能一條一條順序抓取，沒辦法批量併發一塊兒抓取，這簡直就是浪費了nodejs非阻塞機制。異步

好，變通處理添加循環多處理機制。學習

方法3 採用添加間隔器的方式進行批量處理。spa

const DONE = [],
    UNDONE = [],
    FILEDONE = './done.json',
    FILEUNDONE = './undone.json',
    STEP = 20;
let counter = 0, stepi = 0;
const writeResult() =>{
    if(stepi < STEP) return; //只有所有都請求完後才真正執行歸檔功能
    fs.writeFile(FILEDONE, JSON.stringify(DONE), "utf8", ()=>console.log('done saved!'));
    fs.writeFile(FILEUNDONE , JSON.stringify(UNDONE), "utf8", ()=>console.log('undone saved!'));
}
const saveImage = (data, fn)=> {
    fs.writeFile(counter++ + '.jpg', data, fn);
}
const downloadImage = (i)=> {
    if(i >= IMGS.length) {
        stepi++;
        writeResult();
        return;
    }
    http.get(IMGS[i], (res)=> {
            //此處略去了異常處理的內容
         res.setEncoding="binary";
            let dt = '';
            res.on('data', (ck)=> dt+=ck;)
            res.on('end',()=> {
                saveImage(dt, ()=> {
                    DONE.push(IMGS[i]);
                    downloadImag(i + STEP);
                });
            })
        }).on('error',()=>{
            downloadImag(i + STEP);
            UNDONE.push(IMGS[i])
        });
}

for(let i = 0; i < STEP; i++) { downloadImage(i); }//循環step個請求

方法4. 利用promise對象（暫時還未成功，還在想辦法，有想法的同窗能夠貼出來你們借鑑一下）code

let promise_imgs = IMGS.map(v => {
    new Promise((resolve, reject)=> {
        setTimeout(()=>{
            http.get(v, (res)=> {
                //blablabla
                res.on('end', ()=>
                    saveImage(dt, ()=>resolve())
                })
            })
        }, 0);
    });
});

Promise.all(promise_imgs).then(writeResult);//這裏仍是會發生和方法1同樣的問題，多回調異步併發形成網絡連接數的用盡，

你們有什麼好的方法快快貼出來吧！