Mongoose QueryStream.pause() not pausing?

Mongoose QueryStream.pause() not pausing?

本文关键字：not pausing QueryStream pause Mongoose      更新时间：2023-09-26

node v.4.2.3 和 mongoose v.4.3.6

我必须遍历大型（>10k 个文档）集合，并处理每个文档。

在阅读有关如何处理此类迭代的文档时，我偶然发现了QueryStream，我认为它可以解决我的所有问题。

function progress(total, t, current) {
    process.stdout.clearLine();  // clear current text
    process.stdout.write(Math.round(t / total * 100) + '% ' + t + ' / ' + total + ' ' + current);
    process.stdout.cursorTo(0);
}
function loadBalance(current, stream) {
    if(!stream.paused && current > 50) {
        log('DEBUG', 'loadBalance', 'pause');
        stream.pause();
    } else if (stream.paused && current < 10) {
        log('DEBUG', 'loadBalance', 'resume');
        stream.resume();
    }
}
var total = 0,
    error = 0,
    goods = 0,
    current = 0;
stream = Raw.find().stream();
stream.on('data', function (doc) {
    heavyProcess(doc, function (err, refined) {
        current = current + 1;
        loadBalance(current, stream);
        printP(total, goods + error, current);
        if(err) {
            error = error + 1;
            current = current - 1;
            loadBalance(current, stream);
        } else {
            new Pure(refined).save(function (err) {
                if(err) {
                    error = error + 1;
                    current = current - 1;
                    loadBalance(current, stream);
                } else {
                    goods = goods + 1;
                    current = current - 1;
                    loadBalance(current, stream);
                }
            });
        }
    });
}).on('error', function (err) {
    log('ERROR', 'stream', err);
}).on('close', function () {
    log('INFO', 'end', goods + ' / ' + total + ' ( ' + (goods/total*100) + '%) OK_');
    log('INFO', 'end', error + ' / ' + total + ' ( ' + (error/total*100) + '%) NOK');
    log('INFO', 'end', (total - goods - error) + ' missing');
});

loadBalance 确实被调用，打印它正在暂停流，但 'data' 事件继续被触发，甚至认为stream.paused返回 true。

我误解了pause()做什么吗？还是我滥用了 QueryStream？

猫鼬查询流是 v1 流。在文档中称为节点 0.8 读取流 （ http://mongoosejs.com/docs/api.html#querystream_QueryStream ）

这意味着暂停事件是"建议性"https://nodejs.org/api/stream.html#stream_compatibility_with_older_node_js_versions

这里的警告意味着在调用暂停后，某些数据事件仍将泄漏低谷。
这与基础流式处理缓存有关，并且是正确的流 v1 行为。
您必须使用调用暂停后生成的任何数据事件。从开发人员的角度来看，这种行为当然不是最佳的，这就是为什么它在流 v2 中被更改的原因 （ https://nodejs.org/en/blog/feature/streams2/）

这是一个与 v2 查询流相关的 mongoogejs 问题，我认为没有任何计划在短期内实现 v2 查询流。
https://github.com/Automattic/mongoose/issues/1907

引用问题，这可能是您的问题的解决方法：

var readStream = (new stream.Readable({ objectMode: true })).wrap(Model.find({}).stream());

因此，真正的问题不在于我发布的代码，而在于模型生成。

我使用了一个新的连接，原始链接到它，也开始纯粹链接到它，但在最后一刻将其链接到默认的猫鼬连接：

db = mongoose.createConnection('mongodb://127.0.0.1/SNCF');          //Creer la connexion a mongodb
db.on('error', console.error.bind(console, 'connection error:'));
db.once('open', function () {                                       //Une fois connecte
    raw = new mongoose.Schema(
        {
            //...
        },
        {
            strict: true,
            collection: 'Raw'
        }
    );
    Raw = db.model('Raw', raw, 'Raw'); //<--- OK
    pure = new mongoose.Schema(
        {
            //...
        },
        {
            strict: true,
            collection: 'Pure'
        }
    );
    Pure = mongoose.model('Pure', pure, 'Pure'); //<-- ERROR
});

因此，没有保存Pure文档，并且在CPU"正常工作"时内存爆炸。

将错误的行更改为Pure = db.model('Pure', pure, 'Pure');解决了问题，我什至根本不需要暂停流。

相关文章：