javascript - Nodejs Async Promise Queue - Stack Overflow

admin管理员组

文章数量:1406060

I need to consume a rate limited API. For example, I can only make 10 API calls in a second, so I will need to wait the end of the current second to make another API call.

To achieve this, I want to make an asynchronous queue that can manage this on its own. The main functionality of it is to let me add a new promise to the queue, and when the promise is resolved the application is notified:

let queue = new Queue()

queue.add(api.get('/somepath')).then(res => { // handle response });

How can I implement this using ordinary Promises?

export class AsyncQueue {

    private queue: Array<Promise<any>>;


    add(promise, fct) {
        this.queue.push(promise);
    }

    resolveNext() {
        this.queue.pop().then({
            // how to resolve the waiting promise in my application
        })
    }

    get length() {
        return this.queue.length
    }

}

I need to consume a rate limited API. For example, I can only make 10 API calls in a second, so I will need to wait the end of the current second to make another API call.

To achieve this, I want to make an asynchronous queue that can manage this on its own. The main functionality of it is to let me add a new promise to the queue, and when the promise is resolved the application is notified:

let queue = new Queue()

queue.add(api.get('/somepath')).then(res => { // handle response });

How can I implement this using ordinary Promises?

export class AsyncQueue {

    private queue: Array<Promise<any>>;


    add(promise, fct) {
        this.queue.push(promise);
    }

    resolveNext() {
        this.queue.pop().then({
            // how to resolve the waiting promise in my application
        })
    }

    get length() {
        return this.queue.length
    }

}

• javascript
• node.js
• asynchronous
• queue
• es6-promise

Share Improve this question Follow edited May 24, 2018 at 0:38 CertainPerformance 372k5555 gold badges352352 silver badges357357 bronze badges asked May 23, 2018 at 23:05 KarimSKarimS 3,9021010 gold badges4444 silver badges6767 bronze badges 6

• That sounds like a job for Observables. – Jared Smith Commented May 23, 2018 at 23:08
• i think it can be done using only promise? – KarimS Commented May 23, 2018 at 23:12
• Looking at your code it looks like api.get() will be called immediately. Shouldn't get be called after it's been pop()ed from the queue? – Mark Commented May 23, 2018 at 23:12
• Why don't you chain the response handling promise to the api call? – pishpish Commented May 23, 2018 at 23:17
• It can be done with a needle and a steady hand, but I generally prefer a battle-tested pubsub with a thriving munity over my own adhoc implementation. – Jared Smith Commented May 23, 2018 at 23:18

 |  Show 1 more ment

2 Answers 2

Sorted by: Reset to default 6

With the current implementation, api.get() will be called immediately when added to the queue. You should add the path instead (or maybe both api.get and the path) and have AsyncQueue initialize the Promise when it's able. Make sure to have add return a Promise that resolves once the API call is done.

For example, in vanilla JS, it could look like this:

const apiGet = () => new Promise(resolve => setTimeout(resolve, 1000));

class AsyncQueue {
  queue = [];
  constructor() {
    setInterval(this.resolveNext.bind(this), 2000);
  }
  add(fn, param) {
    return new Promise(resolve => {
      this.queue.unshift({ fn, param, resolve });
    });
  }
  resolveNext() {
    if (!this.queue.length) return;
    const { fn, param, resolve } = this.queue.pop();
    fn(param).then(resolve);
  }
}


const queue = new AsyncQueue()
console.log('start');
// Will resolve after 2000 + 1000 seconds:
queue.add(apiGet, '/somepath').then(res => {
  console.log('handling response 1');
});
// Will resolve after 4000 + 1000 seconds:
queue.add(apiGet, '/somepath').then(res => {
  console.log('handling response 2');
});

To avoid permanent call to resolveNext(), is it possible to implement like this ?

class AsyncQueue {
  
    /* delayBetween: delay (ms) before calling next item
        */
  constructor( delayBetween) {
        this.queue = [];
        this.id = 0;
        if (delayBetween < 1) {
            delayBetween = 1;
        }
        this.delayBetween = delayBetween;
        this.timer = null;
    // setInterval( this.resolveNext.bind(this), this.delayBetween);
  }
    
  add(fn, param) {
    return new Promise( resolve => {
        // liste inversée : le dernier élément ajouté est au début du tableau
        this.id ++;
        param.queueId = this.id;
        // console.log( `${new Date().yyyymmddhhmmsslll()} > push request: ${JSON.stringify(param)}`);
      this.queue.unshift( { fn, param, resolve } );
        // console.log( `${new Date().yyyymmddhhmmsslll()} > add() > setTimeout...`);
        if (this.timer == null) {
            this.timer = setTimeout( this.resolveNext.bind(this), this.delayBetween);
        }
    });
  }
    
  resolveNext() {
        this.timer = null;
        // console.log( `${new Date().yyyymmddhhmmsslll()} > resolveNext() > called, len: ${this.queue.length}...`);
    if ( ! this.queue.length) return;       
    const { fn, param, resolve } = this.queue.pop();
        // console.log( `${new Date().yyyymmddhhmmsslll()} > pop request: ${JSON.stringify(param)}`);
        // execute fn, and call resolve only when finished
    // fn(param).then(resolve);
    fn(param).then((result) => {
        // console.log( `${new Date().yyyymmddhhmmsslll()} > fn resolved: ${JSON.stringify(result)}`);
        if (this.timer == null) {
            this.timer = setTimeout( this.resolveNext.bind(this), this.delayBetween);
        }
    });
  }
}

本文标签： javascriptNodejs Async Promise QueueStack Overflow