promise 并发 concurrency

promise-concurrency-limiter

/* TYPES */
type Callback = () => void;
type Task<T> = () => Promise<T>;
type Options = {
  concurrency: number
};
/* PROMISE CONCURRENCY LIMITER */
class Limiter {
  /* VARIABLES */
  concurrency: number;
  count: number;
  queue: Set<Callback>;
  /* CONSTRUCTOR */
  constructor ( options: Options ) {
    this.concurrency = options.concurrency;
    this.count = 0;
    this.queue = new Set ();
  }
  /* API */
  add <T> ( fn: Task<T> ): Promise<T> {
    if ( this.count < this.concurrency ) return this.run ( fn );
    return new Promise<T> ( resolve => {
      const callback = () => resolve ( this.run ( fn ) );
      this.queue.add ( callback );
    });
  }
  flush (): void {
    for ( const callback of this.queue ) {
      if ( this.count >= this.concurrency ) break;
      this.queue.delete ( callback );
      callback ();
    }
  }
  run <T> ( fn: Task<T> ): Promise<T> {
    this.count += 1;
    const promise = fn ();
    const cleanup = (): void => {
      this.count -= 1;
      this.flush ();
    };
    promise.then ( cleanup, cleanup );
    return promise;
  }
}
export default Limiter;

使用：

import Limiter from 'promise-concurrency-limiter';
const limiter = new Limiter ({
  concurrency: 2 // Limit the number of simultaneously active promises to 2
});
const somePromiseReturningFunction = async () => { /* ... */ };
limiter.add ( somePromiseReturningFunction ); // First function added, executed immediately
limiter.add ( somePromiseReturningFunction ); // Second function added, executed immediately
limiter.add ( somePromiseReturningFunction ); // Third function added, executed immediately only if one of the 2 available slots got freed, deferred otherwise

参考

1. promise-concurrency-limiter