Flutter实现CombineExecutor进行多个异步分组监听，监听第一个异步执行的开始和最后一个异步执行结束时机。

傷城~ 2024-03-08 08:05 64阅读 0赞

**1.场景**

我们在调用接口时，很多时候会同时调用多个接口，接口都是异步执行，我们很难知道调用的多个接口哪个会最后执行完成，我们有时候需要对最后一个接口执行完成的时机监听，所以基于该需求，设计了CombineExecutor，对类似的需求进行监听。

**2.代码**

group\_key.dart

///合并执行分类
    class GroupKey {
      ///是否需要监听，不需要监听，则不会执行监听回调
      final bool isMonitor;
    
      GroupKey({this.isMonitor = true});
    }

executor.dart

import 'group_key.dart';
    
    ///执行者
    ///开始一个无限循环的执行进程，等待事件默认50毫秒
    class Executor {
      final GroupKey key;
    
      ///延迟时间。
      ///进程执行的快慢，单位毫秒ms，时间越短，反应越灵敏，
      ///但是消耗的新能越多，不能设置为0，否则会卡住进程。
      ///默认延迟50ms。
      final int? delayed;
      bool _stop = true;
      Function(GroupKey key)? _stopCallback;
      Function(GroupKey key)? _startCallback;
    
      Executor(this.key, {this.delayed});
    
      ///开始执行[Executor]
      ///[callback]会循环调用
      start({Function(GroupKey key)? callback}) async {
        _stop = false;
        _startCallback = callback ?? _startCallback;
        while (!_stop) {
          _startCallback?.call(key);
          await Future.delayed(Duration(milliseconds: delayed ?? 50));
        }
        _stopCallback?.call(key);
      }
    
      ///结束执行[Executor]
      ///[callback]只会在进程结束时执行一次
      stop({Function(GroupKey key)? callback}) {
        _stop = true;
        _stopCallback = callback ?? _stopCallback;
      }
    
      ///是否已启动
      bool isStart() {
        return !_stop;
      }
    
      @override
      int get hashCode => key.hashCode;
    
      @override
      bool operator ==(Object other) =>
          other is! Executor ? false : key == other.key;
    }

monitor.dart

import 'package:kq_flutter_widgets/utils/str_util.dart';
    
    ///合并执行状态持有
    class Monitor {
      dynamic extra;
      bool _isStart = false;
      bool _isFinish = false;
      bool _isError = false;
    
      Monitor({this.extra});
    
      ///该方法接口开始调用时调用
      @Deprecated("只需要监听完成，不需要监听开始，创建即开始")
      onStart() {
        _isStart = true;
      }
    
      ///该方法接口调用完成时调用
      onFinish() {
        _isFinish = true;
      }
    
      ///该方法接口出错或者请求失败时调用
      onError() {
        _isError = true;
      }
    
      ///该接口是否已开始调用
      @Deprecated("只需要监听完成，不需要监听开始，创建即开始")
      bool isStart() {
        return _isStart;
      }
    
      ///该接口是否已完成调用
      bool isFinish() {
        return _isFinish;
      }
    
      ///该接口是否调用出错
      bool isError() {
        return _isError;
      }
    
      ///获取额外数据
      T? getExtra<T>() {
        return StrUtil.getValue(extra);
      }
    
      ///重置，以便复用
      reset() {
        _isStart = false;
        _isFinish = false;
        _isError = false;
      }
    }

str\_util.dart

/// 字符串辅助类
    class StrUtil {
      ///类型判断
      static T? getValue<T>(var value) {
        if (value == null) {
          return null;
        } else if (T == bool) {
          return (value == "1" || value == "true" || value is bool) as T;
        } else if (T == String) {
          return value as T;
        } else if (T == int) {
          return int.parse(value) as T;
        } else if (T == double) {
          return double.parse(value) as T;
        } else {
          return value;
        }
      }
    }

combine\_executor.dart

import 'package:kq_flutter_widgets/utils/ex/kq_ex.dart';
    
    import 'core/executor.dart';
    import 'core/group_key.dart';
    import 'core/monitor.dart';
    
    ///合并执行代码，主要用到接口调用上，
    ///只监测接口执行过程，不涉及接口回调参数等处理。
    ///以最开始执行的接口开始回调[onStart]方法，
    ///以最后执行完成的接口回调[onFinish]方法。
    ///当第一个回调开始了，并已回调完成了，表示整个接口执行完毕，
    ///当第一个接口执行完毕后，还没开始执行第二个接口，则即使他们有共同的[GroupKey]，
    ///他们也不能在一个处理周期中处理，我们把一个同一个[GroupKey]下执行的[onStart]和[onFinish]，
    ///表示一个处理周期。
    class CombineExecutor {
      ///执行的对象保存
      final Map<GroupKey, List<Monitor>> _combines = {};
    
      ///Executor 保存
      final List<Executor> _executors = [];
    
      final Function(GroupKey key)? onStart;
      final Function(GroupKey key)? onFinish;
    
      CombineExecutor({this.onStart, this.onFinish});
    
      _executor(GroupKey key) {
        Executor executor = Executor(key);
        if (!_executors.contains(executor)) {
          _executors.add(executor);
          onStart?.call(key);
    
          executor.start(callback: (key) {
            List<Monitor> combines = _getCombines(key);
            bool flag = true;
            for (Monitor combineMonitor in combines) {
              if (!combineMonitor.isFinish() && !combineMonitor.isError()) {
                flag = false;
                break;
              }
            }
    
            //表示最后一个都已执行完成
            if (flag) {
              executor.stop(callback: (key) {
                onFinish?.call(key);
                _clearCombine(key);
                _executors.remove(executor);
              });
            }
          });
        }
      }
    
      ///停止，在退出界面时调用
      stop() {
        for (Executor executor in _executors) {
          executor.stop();
        }
        _executors.clear();
        _clearAllCombine();
      }
    
      ///获取合并执行观察者，
      ///设置到请求逻辑中。
      Monitor getCombine(GroupKey key) {
        Monitor combineMonitor = Monitor();
        _addCombine(key, combineMonitor);
        _executor(key);
        return combineMonitor;
      }
    
      ///新增一个CombineMonitor
      _addCombine(GroupKey key, Monitor combine) {
        if (key.isMonitor) {
          if (_combines.containsKey(key)) {
            List<Monitor>? combines = _combines[key];
            combines ??= [];
            if (!combines.contains(combine)) {
              combines.add(combine);
            }
          } else {
            _combines.putIfAbsent(key, () => [combine]);
          }
        }
      }
    
      List<Monitor> _getCombines(GroupKey key) {
        if (_isEmptyCombine(key)) {
          return [];
        } else {
          return _combines[key]!;
        }
      }
    
      ///CombineMonitor是否为空
      _isEmptyCombine(GroupKey key) {
        return !_combines.containsKey(key) || _combines[key].isNullOrEmpty;
      }
    
      _clearCombine(GroupKey key) {
        _combines.remove(key);
      }
    
      ///清除全部的CombineMonitor
      _clearAllCombine() {
        _combines.clear();
      }
    }
    
    ///测试
    class Test {
      test() {
        ///创建一个GroupKey，改key可用于一组需要调用的接口上
        GroupKey groupKey = GroupKey();
    
        ///创建对象
        CombineExecutor executor = CombineExecutor(
          onStart: (key) {
            ///print("执行了onStart");
          },
          onFinish: (key) {
            if (key == groupKey) {
              ///print("执行了onFinish");
            }
          },
        );
    
        ///获取CombineMonitor 传入到接口调用中
        Monitor monitor = executor.getCombine(groupKey);
    
        ///模拟异步对Monitor进行操作
        Future.delayed(const Duration(seconds: 2), () {
          monitor.onFinish();
        });
    
        ///退出界面
        executor.stop();
      }
    }

3.使用

![514f35dbaae243a5b287d680759ae7f9.png][]

[514f35dbaae243a5b287d680759ae7f9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/08/ad2ebf0877084d70adfe7b0d4068afa4.png