🎯 Build an Async Task Queue in Dart

1 Define the generic task queue shell

The queue is generic over T, the result type each task produces, so it can be reused for any kind of async work.

// bin/task_queue.dart
import 'dart:async';
import 'dart:collection';

class TaskQueue<T> {
  final int maxConcurrent;
  final Queue<Future<T> Function()> _pending = Queue();
  final StreamController<String> _progressController = StreamController.broadcast();
  int _running = 0;

  TaskQueue({this.maxConcurrent = 2});

  Stream<String> get progress => _progressController.stream;

  // step 2 and 3 methods go here
}

Using a StreamController.broadcast() lets multiple listeners (e.g. a progress bar and a logger) observe queue events independently.

2 Add a task, returning a Future for its result

Each call to add returns immediately with a Future<T> the caller can await, while the actual work is deferred until a concurrency slot opens up.

  Future<T> add(Future<T> Function() task) {
    final completer = Completer<T>();

    _pending.add(() async {
      try {
        final result = await task();
        completer.complete(result);
        return result;
      } catch (e) {
        completer.completeError(e);
        rethrow;
      }
    });

    _tryStartNext();
    return completer.future;
  }

A Completer lets us hand back a Future to the caller now, while controlling exactly when and how it completes later — the bridge between the queue's internal scheduling and the caller's await.

3 Implement the concurrency-limited scheduler

  void _tryStartNext() {
    if (_running >= maxConcurrent || _pending.isEmpty) return;

    final job = _pending.removeFirst();
    _running++;
    _progressController.add('started (running: $_running, queued: ${_pending.length})');

    job().whenComplete(() {
      _running--;
      _progressController.add('finished (running: $_running, queued: ${_pending.length})');
      _tryStartNext(); // pull the next queued job into the freed slot
    });
  }

  Future<void> close() async {
    await _progressController.close();
  }

whenComplete runs regardless of success or failure, guaranteeing _running is always decremented and the next job started — without it, a single failed task would permanently stall the queue.

4 Use the queue

Future<String> simulateDownload(String name, int seconds) async {
  await Future.delayed(Duration(seconds: seconds));
  return 'Downloaded $name';
}

Future<void> main() async {
  final queue = TaskQueue<String>(maxConcurrent: 2);

  final subscription = queue.progress.listen((event) => print('[queue] $event'));

  final futures = [
    queue.add(() => simulateDownload('file1.zip', 2)),
    queue.add(() => simulateDownload('file2.zip', 1)),
    queue.add(() => simulateDownload('file3.zip', 1)),
    queue.add(() => simulateDownload('file4.zip', 2)),
  ];

  final results = await Future.wait(futures);
  for (final r in results) {
    print(r);
  }

  await subscription.cancel();
  await queue.close();
}

5 Run it

dart run bin/task_queue.dart

With maxConcurrent: 2, you should see only two "started" events at a time in the progress log, even though four tasks were added — the third and fourth only start once earlier ones finish and free up a slot.