线程管理

task_management.dart

import 'dart:async';

class TaskManagement<T> {
  int _parallelCount;

  /// 最大线程数
  int get parallelCount => _parallelCount;
  set parallelCount(int value) {
    if (value <= 0) throw Exception('parallelCount must be greater than 0');
    _parallelCount = value;
    _run();
  }

  int _runningCount = 0;

  /// 当前线程数
  int get runningCount => _runningCount;

  final _tasks = <(Completer<T>, Future<T> Function())>[];

  var _isRunning = true;

  /// 是否运行
  bool get isRunning => _isRunning;

  Completer<void>? _stopCompleter;

  TaskManagement(this._parallelCount);

  /// 添加任务
  Future<T> add(Future<T> Function() task) {
    final completer = Completer<T>();
    _tasks.add((completer, task));
    _run();
    return completer.future;
  }

  /// 暂停所有任务
  Future<void> stop() {
    if (!_isRunning) return Future.value();
    _isRunning = false;
    if (_runningCount == 0) return Future.value();
    _stopCompleter = Completer<void>();
    return _stopCompleter!.future;
  }

  /// 启动所有任务
  void start() {
    if (_isRunning) return;
    _isRunning = true;
    _run();
  }

  void _run() {
    while (_isRunning && _runningCount < _parallelCount && _tasks.isNotEmpty) {
      final (completer, task) = _tasks.removeAt(0);
      _runningCount++;
      task()
          .then(completer.complete)
          .onError(completer.completeError)
          .then((_) {
        Future.microtask(() {
          _runningCount--;
          _run();
        });
      });
    }
    if (_runningCount == 0) {
      _stopCompleter?.complete();
      _stopCompleter = null;
    }
  }
}