Dart：如何在异步函数中管理并发

Question

我真的很喜欢 Dart 中的异步/等待模式。 它允许我编写可读的方法。

但是，有几件事是有问题的，特别是一件，我根本不知道要管理什么。

问题在于，在一个方法中使用 async 和多个 await，我们在方法中引入了并发。 例如如果我有一个方法：

Future<int> foo(int value) async {
await foo2();
await foo3();
await foo4();
int ret = foo5(value);
return ret;
}

嗯，这是一个非常简单的例子。 这里的问题是，对于每个等待，该方法都被放入事件循环中。 没关系，当您理解它时，但这不会阻止您的应用程序在重新调整值之前再次调用该方法。

考虑该方法是否管理类实例共有的数据，而不是方法本身。

所以，我尝试了以下解决方案：

bool isWorking = false;

Future<int> foo(int value) async {
if (isWorking) return foo(value);
isWorking = true;

await foo2();
await foo3();
await foo4();
int ret = foo5(value);

isWorking = False;

return ret;
}

据我了解，调用future方法会立即将其放入事件循环中，所以我认为该方法的并发调用的执行延迟到第一个结束。 但不是这样，程序进入死循环。

谁能给我一个解释和解决这个问题的方法？

编辑： 一般来说，我认为像在其他语言中一样，拥有一个 synchronized 关键字可能会很有趣，其含义是如果第二次调用该方法将等到第一次结束。 比如：

Future<int> foo(int value) async synchronized {

编辑 2：

我真的很兴奋，因为我想我找到了解决这个问题的方法，这个问题我解决了很长时间。感谢 Argenti，特别是 Alexandre，他们给了我解决方案。我只是简单地重组了解决方案以便于重用（至少对我而言），并在此处发布了我创建的类以及如何为可能需要它的人使用它的示例（尝试自担风险;-)）。 我使用了一个 mixin，因为我觉得它很实用，但是如果你愿意，你可以单独使用 Locker 类。

myClass extends Object with LockManager {

  Locker locker = LockManager.getLocker();

  Future<int> foo(int value) async {

   _recall() {
      return foo(value);
   } 

   if (locker.locked) {
     return await locker.waitLock();
   }
   locker.setFunction(_recall);
   locker.lock();

   await foo2();
   await foo3();
   await foo4();
   int ret = foo5(value);

   locker.unlock();

   return ret;
  }
}

班级是：

import 'dart:async';

class LockManager {

static Locker getLocker() => new Locker();

}

class Locker {

  Future<Null> _isWorking = null;
  Completer<Null> completer;
  Function _function;
  bool get locked => _isWorking != null;

  lock() {
    completer = new Completer();
    _isWorking = completer.future;
  }

  unlock() {
    completer.complete();
    _isWorking = null;
  }

  waitLock() async {
      await _isWorking;
      return _function();
  }

  setFunction(Function fun) {
    if (_function == null) _function = fun;
  }

}

我以这种方式构造了代码，以便您可以在类中的多个方法中轻松使用它。在这种情况下，每个方法都需要一个 Locker 实例。 希望对你有用。

Answer 1

您可以使用Future 和Completer 而不是布尔值来实现您想要的：

Future<Null> isWorking = null;

Future<int> foo(int value) async {
  if (isWorking != null) {
    await isWorking; // wait for future complete
    return foo(value);
  }

  // lock
  var completer = new Completer<Null>();
  isWorking = completer.future;

  await foo2();
  await foo3();
  await foo4();
  int ret = foo5(value);

  // unlock
  completer.complete();
  isWorking = null;

  return ret;
}

方法第一次调用isWorking 是null，不进入if 部分并创建isWorking 作为将在方法结束时完成的Future。如果在第一次调用完成 Future isWorking 之前对 foo 进行了其他调用，则此调用进入 if 部分并等待 Future isWorking 完成。这对于可以在第一次调用完成之前完成的所有调用都是相同的。一旦第一个呼叫完成（并且isWorking 设置为null），等待的呼叫就会收到通知，他们将再次呼叫foo。其中一个将输入foo 作为第一个呼叫，并且将完成相同的工作流程。

请参阅https://dartpad.dartlang.org/dceafcb4e6349acf770b67c0e816e9a7 以更好地了解工作流程。

Answer 2

答案很好，这只是防止异步操作交错的“互斥锁”的另一种实现。

class AsyncMutex {
  Future _next = new Future.value(null);
  /// Request [operation] to be run exclusively.
  ///
  /// Waits for all previously requested operations to complete,
  /// then runs the operation and completes the returned future with the
  /// result.
  Future<T> run<T>(Future<T> operation()) {
    var completer = new Completer<T>();
    _next.whenComplete(() {
      completer.complete(new Future<T>.sync(operation));
    });
    return _next = completer.future;
  }
}

它没有很多功能，但很简短，希望可以理解。

Answer 3

现在是2021年，我们可以使用synchronized 3.0.0

var lock = new Lock();
Future<int> foo(int value) async {
    int ret;
    await lock.synchronized(() async {
        await foo2();
        await foo3();
        await foo4();
        ret = foo5(value);
    }
    return ret;
}

Answer 4

我猜真正需要的是实现并发管理原语（如锁、互斥体和信号量）的 Dart 库。

我最近使用Pool 包来有效地实现互斥锁以防止对资源的“并发”访问。 （这是“扔掉”代码，所以请不要将其视为高质量的解决方案。）

稍微简化示例：

final Pool pool = new Pool(1, timeout: new Duration(...));

Future<Null> foo(thing, ...) async {
  PoolResource rp = await pool.request();

  await foo1();
  await foo2();
  ...

  rp.release();
}

在调用 foo() 中的异步函数之前从池中请求资源可确保当同时调用多个 foo() 时，对 foo1() 和 foo2() 的调用不会被错误地“交错”。

编辑：

似乎有几个包提供互斥锁：https://www.google.com/search?q=dart+pub+mutex。

Answer 5

异步和并发是两个独立的主题。上面的代码没有什么并发的，都是串行执行的。当 isWorking 为真时，您拥有的代码将继续插入更多 foo(value) 事件 - 它永远不会完成。

您正在寻找的工具是Completer<T>。有问题的特定代码 sn-p 很难辨别，所以我再举一个例子。假设您有共享数据库连接的代码。打开数据库连接是一个异步操作。当一种方法要求建立数据库连接时，它会等待打开完成。在等待期间，另一种方法要求建立数据库连接。期望的结果是只打开一个数据库连接，并将一个数据库连接返回给两个调用者：

Connection connection;
Completer<Connection> connectionCompleter;
bool get isConnecting => connectionCompleter != null;

Future<Connection> getDatabaseConnection() {
  if (isConnecting) {
    return connectionCompleter.future;
  } 

  connectionCompleter = new Completer<Connection();

  connect().then((c) {
    connection = c;
    connectionCompleter.complete(c);
    connectionCompleter = null;
  });

  return connectionCompleter.future;
}

第一次调用getDatabaseConnection 时，会调用connect 函数，并将完成者的未来返回给调用者。假设在connect 完成之前再次调用getDatabaseConnection，则返回相同的完成者的未来，但不会再次调用connect。

当connect 完成时，完成者的事件被添加到事件队列中。这会触发从getDatabaseConnection 返回的两个期货按顺序完成，并且connection 将有效。

注意getDatabaseConnection 甚至不是一个异步方法，它只是返回一个Future。它可能是异步的，但它只会浪费 CPU 周期并使您的调用堆栈变得丑陋。