Node.js 在多次 setTimeout 循环调用时挂起

Question

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

class 还使用了我编写的 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”之前停止 RIGHT。

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

一如既往，非常感谢！

Answer 1

找到问题了！事实证明 Node.js 不喜欢你将小数传递给 setTimeout！四舍五入小数点解决了这个问题。

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

如果此代码对任何人有用，请告诉我。如果是，我可以 post 一个 github 当它全部完成时。不过现在它完全可以工作了。

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

Node.js 在多次 setTimeout 循环调用时挂起

Node.js hangs on multiple setTimeout looping calls

javascript

throttling

freeze

settimeout

node.js