serial_queue 1.0.1

Convert concurrent business processes into serial sequential execution of each business process.

Features

Convert concurrent business processes into serial sequential execution of each business process.

Getting started

Install the package

Usage

Example1:

import 'package:serial_queue/serial_queue.dart';
import 'package:serial_queue/sleep.dart';

int i = 0;

void main() async {
  SerialQueue queue = SerialQueue()..startQueue();

  Future.delayed(const Duration(milliseconds: 20), () async {
    Task addTask = Task<int, List<int>>.create(taskHandler: add, params: [1, 2, 3, 4, 5, 6, 7, 8, 9]);
    queue.addTask(addTask);
    var sum = await addTask.future;
  });

  Future.delayed(const Duration(milliseconds: 20), () async {
    Task incrementTask = Task<void, String>.create(taskHandler: increment, params: 'D');
    queue.addTask(incrementTask);
    await incrementTask.future;
  });

  Future.delayed(const Duration(milliseconds: 20), () async {
    Task showTask = Task<void, String?>.create(taskHandler: show, params: 'text content');
    queue.addTask(showTask);
    await showTask.future;
  });
  await sleep(1000);
}

int add({List<int>? params})  {
  var sum = params!.reduce((value, element) => value + element);
  return sum;
}

void show({String? params}) async {
  print('$params');
  return;
}

Future<int> increment({String? params}) async {
  int a = i;
  await sleep(3);
  a++;
  i = a;
  print('正在处理 increment 任务----$params--$i');
  return i;
}

Example2:

import 'package:flutter_test/flutter_test.dart';
import 'package:serial_queue/serial_queue.dart';
import 'package:serial_queue/sleep.dart';


/// 100个人在5台atm机上存钱
void main() async {
  bool useQueue = false;
  Bank bank = Bank(name: 'ABC', useQueue: useQueue);

  var team1 = bank.persons.sublist(0, 20);
  var team2 = bank.persons.sublist(20, 40);
  var team3 = bank.persons.sublist(40, 60);
  var team4 = bank.persons.sublist(60, 80);
  var team5 = bank.persons.sublist(80, 100);

  Future.delayed(const Duration(), () async {
    for (var person in team1) {
      await bank.atms[0]?.deposit(person!.accountNo, 100);
      await sleep(30);
    }
  });
  Future.delayed(const Duration(), () async {
    for (var person in team2) {
      await bank.atms[1]?.deposit(person!.accountNo, 100);
      await sleep(2);
    }
  });
  Future.delayed(const Duration(), () async {
    for (var person in team3) {
      await bank.atms[2]?.deposit(person!.accountNo, 100);
      await sleep(2);
    }
  });
  Future.delayed(const Duration(), () async {
    for (var person in team4) {
      await bank.atms[3]?.deposit(person!.accountNo, 100);
      await sleep(2);
    }
  });
  Future.delayed(const Duration(), () async {
    for (var person in team5) {
      await bank.atms[4]?.deposit(person!.accountNo, 100);
      await sleep(2);
    }
  });
  // Future.delayed(const Duration(milliseconds: 200), () async {
  //   bank.atms[0]!.hasError = true;
  //   await sleep(200);
  //   bank.atms[0]!.hasError = false;
  // });
  await sleep(3000);
  await bank.close();
}

///
class Person {
  Person({required this.accountNo});
  String accountNo;
  double money = 0;
  void look() {}
}

///
class ATM {
  Bank bank;
  String id;
  double money = 0;
  bool hasError = false;

  ATM({required this.bank, required this.id});

  Future<bool> deposit(String accountNo, double money) async {
    if (hasError) {
      return await bank.atmTimeout();
    }
    return await bank.atmDeposit(id, accountNo, money);
  }

  void look() {
    print('ATM id:$id   money=$money');
  }
}

///
class Bank {
  double total = 0;
  var persons = <Person?>[];
  var atms = <ATM?>[];
  String name;
  bool useQueue = false;
  late SerialQueue queue;
  int orderId = 0;

  ///
  Bank({required this.name, this.useQueue = false}) {
    queue = SerialQueue(log: true)..startQueue();
    for (int i = 0; i < 5; i++) {
      atms.add(ATM(bank: this, id: 'id-$i'));
    }
    for (int i = 0; i < 100; i++) {
      persons.add(Person(accountNo: 'N@$i'));
    }
  }

  Future<bool> atmTimeout() {
    var task = Task<bool, String>.create(
      taskHandler: ({String? params}) {
        print(params);
        return false;
      },
      params: '-----------> ATM故障!!!',
    );
    queue.addTask(task);
    return task.future.onError((error, stackTrace){
      print('$error');
      return false;
    });
  }

  ///
  Future<bool> atmDeposit(String fromAtmId, String accountNo, double money) async {
    if (useQueue) {
      var task = Task<bool, DepositInfo>.create(taskHandler: _atmDeposit, params: DepositInfo(fromAtmId: fromAtmId, accountNo: accountNo, money: money));
      queue.addTask(task);
      var r = await task.future.onError((error, stackTrace){
        print('$error');
        return false;
      });
      return r;
    } else {
      return _atmDeposit(params: DepositInfo(fromAtmId: fromAtmId, accountNo: accountNo, money: money));
    }
  }

  ///
  Future<bool> _atmDeposit({DepositInfo? params}) async {
    print('orderId:${orderId++}     fromAtmId:${params!.fromAtmId}  ${params.accountNo}  ${params.money}');
    await sleep(2);
    var atm = atms.firstWhere((atm) => atm?.id == params.fromAtmId, orElse: () => null);
    if (atm == null) {
      return false;
    }
    var person = persons.firstWhere((p) => p?.accountNo == params.accountNo, orElse: () => null);
    if (person == null) {
      return false;
    }
    double t = total;
    double atmT = atm.money;
    await sleep(2);
    person.money += params.money;
    atmT += params.money;
    t += params.money;
    await sleep(1);
    total = t;
    atm.money = atmT;
    return true;
  }

  ///
  void look() {
    print('Bank name:$name   total=$total');
    for (var atm in atms) {
      atm!.look();
    }
    for (var p in persons) {
      p!.look();
    }
  }

  Future<void> close() async {
    await queue.dispose(() {
      print('---------------------银行结算信息---------------------');
      look();
    });
  }
}

class DepositInfo {
  String fromAtmId;
  String accountNo;
  double money;

  DepositInfo({required this.fromAtmId, required this.accountNo, required this.money});
}