如何在JavaScript中实现带并发控制的任务队列（Job queue）？

Got it, let's break this down. A job queue that handles async functions with concurrency control (and runs tasks one by one by default) is perfect for managing work like API calls, file operations, or any async task where you don't want to flood your system with too many concurrent operations.

Core Concepts #

First, let's outline the key pieces we need to make this work:

• A task queue: Stores jobs waiting to execute when a slot opens up.
• Active task counter: Tracks how many jobs are currently running (to enforce our concurrency limit).
• Execution logic: When a job is added, if we haven't hit our concurrency limit, run it immediately. Otherwise, add it to the queue. When a job finishes, pull the next one from the queue and run it.

Complete Implementation #

Here's a robust, reusable class that fits your needs—it supports async functions, returns results, and lets you set a concurrency limit (default is 1 for sequential execution):

class JobQueue {
  constructor(maxConcurrency = 1) {
    this.queue = [];
    this.activeCount = 0;
    this.maxConcurrency = maxConcurrency;
  }

  // Add a job to the queue, returns a promise that resolves with the job's result
  add(job) {
    return new Promise((resolve, reject) => {
      // Wrap the job with its resolve/reject handlers to pass back results
      this.queue.push({
        job,
        resolve,
        reject
      });
      // Try to run the next job right after adding it
      this._runNext();
    });
  }

  // Private method to run the next job in the queue if possible
  _runNext() {
    // Exit if we're at max concurrency or the queue is empty
    if (this.activeCount >= this.maxConcurrency || this.queue.length === 0) {
      return;
    }

    // Pull the next job from the queue
    const { job, resolve, reject } = this.queue.shift();
    this.activeCount++;

    // Execute the job—handle both sync and async functions uniformly
    Promise.resolve(job())
      .then(result => {
        resolve(result);
      })
      .catch(error => {
        reject(error);
      })
      .finally(() => {
        // When the job finishes, free up a slot and run the next job
        this.activeCount--;
        this._runNext();
      });
  }
}

How It Works #

Let's walk through the key parts:

• Constructor: Initializes the queue, active task counter, and sets the max concurrency (default 1 for one-by-one execution).
• add() Method: Takes a function (sync or async) and returns a Promise. We wrap the job with its resolve/reject handlers so we can pass the job's result back to the caller, then trigger _runNext() to try executing the job immediately.
• _runNext() Method: Checks if we have available slots. If yes, it pulls the next job from the queue, increments the active count, and runs the job. Using Promise.resolve(job()) ensures we handle both sync and async jobs the same way. Once the job finishes (success or failure), we free up a slot and call _runNext() again to process the next job in line.

Example Usage #

Let's test this with mock async tasks to see it in action:

// Create a queue that runs 1 task at a time (default behavior)
const queue = new JobQueue();

// Mock async task that resolves after a delay
const createTask = (id, delay) => {
  return () => new Promise(resolve => {
    console.log(`Starting task ${id}`);
    setTimeout(() => {
      const result = `Result from task ${id}`;
      console.log(`Finished task ${id}`);
      resolve(result);
    }, delay);
  });
};

// Add tasks to the queue
queue.add(createTask(1, 2000)).then(result => console.log(result));
queue.add(createTask(2, 1000)).then(result => console.log(result));
queue.add(createTask(3, 500)).then(result => console.log(result));

// Output will be:
// Starting task 1
// Finished task 1
// Result from task 1
// Starting task 2
// Finished task 2
// Result from task 2
// Starting task 3
// Finished task 3
// Result from task 3

If you want to run multiple tasks concurrently, just pass a higher maxConcurrency value:

// Queue that runs 2 tasks at once
const concurrentQueue = new JobQueue(2);
concurrentQueue.add(createTask(1, 2000)).then(console.log);
concurrentQueue.add(createTask(2, 1000)).then(console.log);
concurrentQueue.add(createTask(3, 500)).then(console.log);

// Output will be:
// Starting task 1
// Starting task 2
// Finished task 2
// Result from task 2
// Starting task 3
// Finished task 3
// Result from task 3
// Finished task 1
// Result from task 1

Key Features #

• Handles both sync and async functions seamlessly.
• Returns a Promise for each job so you can handle results/errors individually.
• Enforces concurrency limits to prevent overwhelming your runtime.
• Automatically processes the queue as jobs finish.

内容的提问来源于stack exchange，提问作者Rishab Jain