約束の中のいくつかのものが実行されていません

Question

大きなCSVファイルを読み込んで解析しようとしていましたが、行ごとに非同期計算をいくつか行い、だから、私はPromise pを作り、多くの.then(xxx)を連鎖しようとし、csvの最後には.then(yyy)という最後のカウントを出力する。

ただし、この数値は加算されません。しかし、私がp = p.then(xxx)とp = p.then(yyy)を行うと、その数は（より小さいcsvファイルの場合）加算されますが、（大容​​量のcsvファイルの場合）時にはメモリリークに直面します。

私が間違っていたことはありますか？

var fs = require('fs') 
const csv = require('fast-csv'); 
var Promise = require('bluebird') 
var count = 0; 
var actual = 0; 
let p = Promise.resolve(); 
const stream = fs.createReadStream(`/Users/ssmlee/Desktop/KingKong_Sims_5M.txt`); 
const csvStream = csv({ 
    delimiter: ';' 
}) 
.on('data', (row) => { 
    count++ 
    if (count % 10000 === 0) { 
     console.log(count) 
     console.log(process.memoryUsage().heapUsed) 
    } 
    p.then(() => { // instead if we do p = p.then(() => it will work correctly 
    return Promise.resolve().delay(5) 
    .then(function() { 
     actual++ 
    }) 
    }); 
}) 
.on('end',() => { 
    p.then(() => { // instead if we do p = p.then(() => it will work correctly 
    console.log(actual); // 4999977 or something like this 
    console.log(count); // 5000000 
    }); 
}); 
stream.pipe(csvStream); 

Answer 1

あなたが遅延してactual数を増やすませんが、決して（離れthenの結果を投げて）約束を待つ場合は、ストリームがすでに起きていないすべての増分で終わるかもしれません。あなたの例では、23コールバックはまだ5msの遅延を待っています。 Btw、これらのすべてを同じに連結するp = Promise.resolve()はあまり意味がありません。ただすぐにすべてを実行できます。

p = p.then(…)を実行している場合、非常に長い約束チェーンを構築します。これはメモリをリークさせるべきではありませんが、多くのメモリを使用します - 5msの遅延はすべて連続して連鎖し、スクリプトは実行するには少なくとも25000秒かかるでしょう。ファイルは最初に読み込まれ、何百万もの約束事が生成された後、次々と解決されます（ガベージコレクションされる可能性があります）。

このような連続的なアプローチを実行するには、おそらくストリームの背圧システムを使用する必要があります。

しかし、あなたは一度にあまりにも多くの生きているの約束を持つ、並列の遅延をも待つことができます。

p = Promise.all([p, Promise.delay(5)]).then(() => { 
    actual++; 
}); 

Answer 2

さてあなたは約束を並列に実行するので、それらをチェーンすることはできません。

allp = []; 
.... 
.on('data', (row) => { 
    ... 
    allp.push(p.then(() => {...})); 
} 
... 
.on('end',() => { 
Promise.all(allp).then(() => {}) 

もちろん、イベントごとに1つのプロミスを作成しています。

最後までに約束を解除する必要がある場合は、これを自分で行う必要があります。

ので、あなたが行うことができますが、唯一その副作用（増加数）で、約束の戻り値に興味があるように思えません

.on('data', (row) => { 
    ... 
    if (allp.length > 50) allp = [Promise.all(allp).then(()=>null)]; 
    allp.push(p.then(() => {...})); 
} 

50個の約束がグループ化されますこうすることで、彼らは解決した後と、彼らは

.then(()=>null)はPromise.allから結果の配列があまりにも破棄されることを保証（...次の50になります）、単一の約束によって置き換えられます。 （代わりに1つのnullの約束はallpになります）

これはPromise.allの実装に依存します。 Promise.allがそれぞれの約束を解決したときに（そしてその結果が利用可能な場合）、これは完璧です。

Promise.allが50の約束をすべて待ってからすべて解放すると、50の各グループに1つの非常に長い実行予定がない限り、これはまだ機能します。

---

延期約束の反パターンを使用することができます。

開始時に1つの延期約束を作成します。値1データ

.on('data', (row) => { 
    ... 
    asyncRunningCount++; 
    p.then(() => {work})) 
    .then(() { 
     asyncRunningCount--; 
     if (asyncRunningCount == 0) resolve(); // no more tasks running 
    }); 
} 

.on('end',() => { 
    asyncRunningCount--; 
    // remove the 1 that was set on start. No more new tasks will be added 
    if (asyncRunningCount == 0) resolve(); // no more tasks running 
    p2.then(() => { all done }) 

上で

var resolve; 
var asyncRunningCount = 1; // start with 1 
var p2 = new Promise(function() { 
    resolve = arguments[0]; 
}); 

起動時にタスクを実行するのカウントを一時的に（端部）には0

に低下した場合、分解されることから、P2を防ぎ1が減分される。すべてのタスクが完了すると、asyncRunningCountは0になります。これは、オン（終了）のデクリメントまたはオン（データ）のデクリメントによって発生します。

p2.thenは、すべてのタスクが終了すると実行されます。

他の約束事はすべて終了します。 実際には（データ）をオンにすると約束は必要ありません。ちょうどあなたの非同期タスクを開始し、非同期タスクが減少asyncRunningCountを行って、そして0

---

をチェックされたときに、データが非常に高速で来る場合、これはまだ約束の多くは並行して開始することを、意味しています。 しかし、あなたが約束を開始しない場合は、着信データを格納する必要があるので、メモリはいずれかの方法で使用されます。