javascript - Node.js 在多个 setTimeout 循环调用上挂起

Question

我正在尝试编写一个类，该类将允许我暂停音频文件的播放。该类接受原始 PCM 数据，并且您向该类提供发送样本 block 的频率。例如，您可以指定每 20 毫秒传递一个 block 。该类还实现了暂停()和恢复()函数。

该类还使用了我编写的 SerialQueue 模块，以确保数据不会在缓冲区上同时被切片和连接。因此，您将在以下代码中看到很多对此的引用。

我遇到的问题是它会多次调用 setTimeout，但最终会在 setTimeout 上随机卡住。我的处理器使用率将立即飙升，并且不会发生任何其他事情。

这是完整的代码加上我的测试代码:

var nopus = require( './lib/node-opus' );
var Speaker = require( 'speaker' );
var fs = require( 'fs' );
var path = require( 'path' );
var SerialQueue = require( './lib/serial-queue' );
var Transform = require( 'stream' ).Transform;
var inherits = require( 'util' ).inherits;

function ThrottlePCM( opts ) {
    // Default to an empty object
    if( !opts )
        opts = {};

    // Pass through the options
    Transform.call( this, opts );

    this.milliseconds = opts.milliseconds | 20;
    this.bitDepth = opts.bitDepth | 16;
    this.channels = opts.channels | 1;
    this.sampleRate = opts.sampleRate | 48000;

    // Set our frame size
    if( this.milliseconds==2.5 )
        this.frameSize = this.sampleRate/400;
    else if( this.milliseconds==5 )
        this.frameSize = this.sampleRate/200;
    else if( this.milliseconds==10 )
        this.frameSize = this.sampleRate/100;
    else if( this.milliseconds==20 )
        this.frameSize = this.sampleRate/50;
    else if( this.milliseconds==40 )
        this.frameSize = this.sampleRate/25;
    else if( this.milliseconds==60 )
        this.frameSize = 3*this.sampleRate/50;
    else
        throw new Error( "Millisecond value is not supported." );

    this.bytesPerBeat = this.frameSize*this.bitDepth/8*this.channels;

    console.log( "Bytes per beat %d.", this.bytesPerBeat );

    this.buffer = null;
    this.queue = new SerialQueue();

    // Taken from TooTallNate
    this.totalBytes = 0;
    this.startTime = Date.now();
    this.pauseTime = null;

    // Can we pass
    this.canPass = true;
    this.paused = false;
    this.flushCallback = null;
}

inherits( ThrottlePCM, Transform );

ThrottlePCM.prototype._transform = function( data, encoding, done ) {
    var that = this;

    this.queue.queue( function() {
        // Append the buffer
        if( that.buffer )
            that.buffer = Buffer.concat( [ that.buffer, data ] );
        else
            that.buffer = data;

        // Nen no tame
        if( that.canPass )
            that.passThrough();
    } );

    // We are ready for more data
    done();
};

ThrottlePCM.prototype.pause = function() {
    this.paused = true;
    this.pauseTime = Date.now();
};

ThrottlePCM.prototype.resume = function() {
    this.paused = false;
    this.startTime+= Date.now()-this.pauseTime;

    console.log( "Difference is %d: %d", Date.now()-this.pauseTime, this.startTime );

    var that = this;

    this.queue.queue( function() {
        that.passThrough();
    } );

};

ThrottlePCM.prototype.passThrough = function() {
    // Are we paused?
    if( this.paused ) {
        this.canPass = true;
        return;
    }

    // No pass now
    this.canPass = false;

    // The rest of us
    var that = this;
    var totalBeats = (Date.now()-this.startTime)/this.milliseconds;
    var expected = totalBeats*this.bytesPerBeat;

    function passMe() {
        console.log( "== Inkasemeen" );
        that.queue.queue( function() {
            if( !that.buffer ) {
                // Should we just flush?
                if( that.flushCallback ) {
                    var callback = that.flushCallback;
                    that.flushCallback = null;

                    console.log( "Antipass" );

                    callback();
                }
                else
                    that.canPass = true; // We can pass now from on timer
                return;
            }

            var output;

            if( that.buffer.length>that.bytesPerBeat ) {
                output = that.buffer.slice( 0, that.bytesPerBeat );
                that.buffer = that.buffer.slice( that.bytesPerBeat );
            }
            else {
                output = that.buffer;
                that.buffer = null;
            }

            that.push( output );
            that.totalBytes+= output.length;

            // Re-call us
            that.passThrough();
        } );
    }

    console.log( "--\nTotal Beats: %d\nTotal Bytes: %d\nExpected: %d\nBytes Per Beat: %d\nMilliseconds: %s", totalBeats, this.totalBytes, expected, this.bytesPerBeat, this.milliseconds );

    if( this.totalBytes>expected ) {
        var remainder = this.totalBytes-expected;
        var sleepTime = remainder/this.bytesPerBeat*this.milliseconds;

        console.log( "++\nSleep time: %d", sleepTime );

        if( sleepTime ) {
            setTimeout( passMe, sleepTime );
        }
        else {
            passMe();
        }
    }
    else {
        console.log( "Bytes are higher by %d (%d-%d)", expected-this.totalBytes, expected, this.totalBytes );
        passMe();
    }
};

ThrottlePCM.prototype._flush = function( done ) {
    console.log( "Flush called." );

    // No action here I don't think
    this.flushCallback = done;

    var that = this;

    this.queue.queue( function() {
        // Show ourselves flushy
        if( that.canPass )
            that.passThrough();
    } );
};

var format = {
    channels: 1,
    bitDepth: 16,
    sampleRate: 48000,
    bitrate: 16000,
    milliseconds: 60
};

var rate = nopus.getFrameSizeFromMilliseconds*format.channels*nopus.binding.sizeof_opus_int16;
var speaker = new Speaker( format );
var decoder = new nopus.Decoder( format );
var throttle = decoder.pipe( new ThrottlePCM( format ) );

throttle.pipe( speaker );

var file = fs.createReadStream( path.join( __dirname, 'files/audio/233' ) );

file.pipe( decoder );

这会产生以下输出:

Bytes per beat 5760.
--
Total Beats: 0.1
Total Bytes: 0
Expected: 576
Bytes Per Beat: 5760
Milliseconds: 60
Bytes are higher by 576 (576-0)
== Inkasemeen
--
Total Beats: 0.15
Total Bytes: 1920
Expected: 864
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 11
== Inkasemeen
--
Total Beats: 0.26666666666666666
Total Bytes: 7680
Expected: 1536
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 64
== Inkasemeen
--
Total Beats: 1.3666666666666667
Total Bytes: 13440
Expected: 7872
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 58
== Inkasemeen
--
Total Beats: 2.3833333333333333
Total Bytes: 19200
Expected: 13728
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 57
== Inkasemeen
--
Total Beats: 3.283333333333333
Total Bytes: 24960
Expected: 18912
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 63
== Inkasemeen
--
Total Beats: 4.35
Total Bytes: 30720
Expected: 25055.999999999996
Bytes Per Beat: 5760
Milliseconds: 60
++
Sleep time: 59.000000000000036

它始终卡在不同的位置。正如您所看到的，它在调用 passMe 函数并打印“== Inkasemeen”之前停止。

我的 Node.js 版本是 v0.10.30。

一如既往，非常感谢!

Answer 1

发现问题了!事实证明，Node.js 不喜欢你向 setTimeout 传递小数!对小数点进行四舍五入解决了问题。

if( sleepTime>0 ) {
    setTimeout( passMe, sleepTime|0 );
}
else {
    passMe();
}

请告诉我此代码是否对任何人有用。如果是的话我可以在全部完成后发布一个 github。不过现在它完全可以工作了。

另请注意，TooTallNate 已经推出了一个 Throttle 模块，它可能可以满足大多数人对流限制的需求 https://github.com/TooTallNate/node-throttle .