Flutter如何保證資料操作原子性詳解

2022-03-04 19:01:23

前言

Flutter 是單執行緒架構，按道理理說，Flutter 不會出現 Java 的多執行緒相關的問題。

但在我使用 Flutter 過程中，卻發現 Flutter 依然會存在資料操作原子性的問題。

其實 Flutter 中存在多執行緒的（Isolate 隔離池），只是 Flutter 中的多執行緒更像 Java 中的多程序，因為 Flutter 中執行緒不能像 Java 一樣，可以兩個執行緒去操作同一個物件。

我們一般將計算任務放在 Flutter 單獨的執行緒中，例如一大段 Json 資料的解析，可以將解析計算放在單獨的執行緒中，然後將解析完後的 Map<String, dynamic> 返回到主執行緒來用。

Flutter單例模式

在 Java 中，我們一般喜歡用單例模式來理解 Java 多執行緒問題。這裡我們也以單例來舉例，我們先來一個正常的：

class FlutterSingleton {
  static FlutterSingleton? _instance;

  /// 將構造方法宣告成私有的
  FlutterSingleton._();

  static FlutterSingleton getInstance() {
    if (_instance == null) {
      _instance = FlutterSingleton._();
    }
    return _instance!;
  }
}

由於 Flutter 是單執行緒架構的, 所以上述程式碼是沒有問題的。

問題範例

但是, 和 Java 不同的是, Flutter 中存在非同步方法。

做 App 開發肯定會涉及到資料持久化，Android 開發應該都熟悉 SharedPreferences，Flutter 中也存在 SharedPreferences 庫，我們就以此來舉例。同樣實現單例模式，只是這次無可避免的需要使用 Flutter 中的非同步：

class SPSingleton {
  static SPSingleton? _instance;

  String? data;

  /// 將構造方法宣告成私有的
  SPSingleton._fromMap(Map<String, dynamic> map) : data = map['data'];

  static Future<SPSingleton> _fromSharedPreferences() async {
    // 模擬從 SharedPreferences 中讀取資料, 並以此來初始化當前物件
    Map<String, String> map = {'data': 'mockData'};
    await Future.delayed(Duration(milliseconds: 10));
    return SPSingleton._fromMap(map);
  }

  static Future<SPSingleton> getInstance() async {
    if (_instance == null) {
      _instance = await SPSingleton._fromSharedPreferences();
    }
    return _instance!;
  }
}

void main() async {
  SPSingleton.getInstance().then((value) {
    print('instance1.hashcode = ${value.hashCode}');
  });
  SPSingleton.getInstance().then((value) {
    print('instance2.hashcode = ${value.hashCode}');
  });
}

執行上面的程式碼，列印紀錄檔如下：

instance1.hashcode = 428834223
instance2.hashcode = 324692380

可以發現，我們兩次呼叫 SPSingleton.getInstance() 方法，分別建立了兩個物件，說明上面的單例模式實現有問題。

我們來分析一下 getInstance() 方法：

static Future<SPSingleton> getInstance() async {
  if (_instance == null) { // 1
    _instance = await SPSingleton._fromSharedPreferences(); //2
  }
  return _instance!;
}

當第一次呼叫 getInstance() 方法時，程式碼在執行到 1 處時，發現 _instance 為 null, 就會進入 if 語句裡面執行 2 處, 並因為 await 關鍵字掛起, 並交出程式碼的執行權, 直到被 await 的 Future 執行完畢，最後將建立的 SPSingleton 物件賦值給 _instance 並返回。

當第二次呼叫 getInstance() 方法時，程式碼在執行到 1 處時，可能會發現 _instance 還是為 null (因為 await SPSingleton._fromSharedPreferences() 需要 10ms 才能返回結果), 然後和第一次呼叫 getInstance() 方法類似, 建立新的 SPSingleton 物件賦值給 _instance。

最後導致兩次呼叫 getInstance() 方法, 分別建立了兩個物件。

解決辦法

問題原因知道了，那麼該怎樣解決這個問題呢？

究其本質，就是 getInstance() 方法的執行不具有原子性，即：在一次 getInstance() 方法執行結束前，不能執行下一次 getInstance() 方法。

幸運的是, 我們可以藉助 Completer 來將非同步操作原子化，下面是藉助 Completer 改造後的程式碼：

import 'dart:async';

class SPSingleton {
  static SPSingleton? _instance;
  static Completer<bool>? _monitor;

  String? data;

  /// 將構造方法宣告成私有的
  SPSingleton._fromMap(Map<String, dynamic> map) : data = map['data'];

  static Future<SPSingleton> _fromSharedPreferences() async {
    // 模擬從 SharedPreferences 中讀取資料, 並以此來初始化當前物件
    Map<String, String> map = {'data': 'mockData'};
    await Future.delayed(Duration(milliseconds: 10));
    return SPSingleton._fromMap(map);
  }

  static Future<SPSingleton> getInstance() async {
    if (_instance == null) {
      if (_monitor == null) {
        _monitor = Completer<bool>();
        _instance = await SPSingleton._fromSharedPreferences();
        _monitor!.complete(true);
      } else {
        // Flutter 的 Future 支援被多次 await
        await _monitor!.future;
        _monitor = null;
      }
    }
    return _instance!;
  }
}

void main() async {
  SPSingleton.getInstance().then((value) {
    print('instance1.hashcode = ${value.hashCode}');
  });
  SPSingleton.getInstance().then((value) {
    print('instance2.hashcode = ${value.hashCode}');
  });
}

我們再次分析一下 getInstance() 方法:

static Future<SPSingleton> getInstance() async {
  if (_instance == null) { // 1
    if (_monitor == null) { // 2
      _monitor = Completer<bool>(); // 3
      _instance = await SPSingleton._fromSharedPreferences(); // 4
      _monitor!.complete(true); // 5
    } else {
      // Flutter 的 Future 支援被多次 await
      await _monitor!.future; //6
      _monitor = null;
    }
  }
  return _instance!; // 7
}

當第一次呼叫 getInstance() 方法時, 1 處和 2 處都會判定為 true, 然後進入執行到 3 處建立一個的 Completer 物件, 然後在 4 的 await 處掛起, 並交出程式碼的執行權, 直到被 await 的 Future 執行完畢。

此時第二次呼叫的 getInstance() 方法開始執行，1 處同樣會判定為 true, 但是到 2 處時會判定為 false, 從而進入到 else, 並因為 6 處的 await 掛起, 並交出程式碼的執行權;

此時, 第一次呼叫 getInstance() 時的 4 處執行完畢, 並執行到 5, 並通過 Completer 通知第二次呼叫的 getInstance() 方法可以等待獲取程式碼執行權了。

最後，兩次呼叫 getInstance() 方法都會返回同一個 SPSingleton 物件，以下是列印紀錄檔：

instance1.hashcode = 786567983
instance2.hashcode = 786567983

由於 Flutter 的 Future 是支援多次 await 的, 所以即便是連續 n 次呼叫 getInstance() 方法, 從第 2 到 n 次呼叫會 await 同一個 Completer.future, 最後也能返回同一個物件。

Flutter任務佇列

雖然我們經常拿單例模式來解釋說明 Java 多執行緒問題，可這並不代表著 Java 只有在單例模式時才有多執行緒問題。

同樣的，也並不代表著 Flutter 只有在單例模式下才有原子操作問題。

問題範例

我們同樣以資料持久化來舉例，只是這次我們以資料庫操作來舉例。

我們在運算元據庫時，經常會有這樣的需求：如果資料庫表中存在這條資料，就更新這條資料，否則就插入這條資料。

為了實現這樣的需求，我們可能會先從資料庫表中查詢資料，查詢到了就更新，沒查詢到就插入，程式碼如下：

class Item {
  int id;
  String data;
  Item({
    required this.id,
    required this.data,
  });
}

class DBTest {
  DBTest._();
  static DBTest instance = DBTest._();
  bool _existsData = false;
  Future<void> insert(String data) async {
    // 模擬資料庫插入操作,10毫秒過後,資料庫中才有資料
    await Future.delayed(Duration(milliseconds: 10));
    _existsData = true;
    print('執行了插入');
  }

  Future<void> update(String data) async {
    // 模擬資料庫更新操作
    await Future.delayed(Duration(milliseconds: 10));
    print('執行了更新');
  }

  Future<Item?> selected(int id) async {
    // 模擬資料庫查詢操作
    await Future.delayed(Duration(milliseconds: 10));
    if (_existsData) {
      // 資料庫中有資料才返回
      return Item(id: 1, data: 'mockData');
    } else {
      // 資料庫沒有資料時,返回null
      return null;
    }
  }

  /// 先從資料庫表中查詢資料，查詢到了就更新，沒查詢到就插入
  Future<void> insertOrUpdate(int id, String data) async {
    Item? item = await selected(id);
    if (item == null) {
      await insert(data);
    } else {
      await update(data);
    }
  }
}

void main() async {
  DBTest.instance.insertOrUpdate(1, 'data');
  DBTest.instance.insertOrUpdate(1, 'data');
}

我們期望的輸出紀錄檔為:

執行了插入
執行了更新

但不幸的是, 輸出的紀錄檔為:

執行了插入
執行了插入

原因也是非同步方法運算元據, 不是原子操作, 導致邏輯異常。

也許我們也可以效仿單例模式的實現，利用 Completer 將 insertOrUpdate() 方法原子化。

但對於資料庫操作是不合適的，因為我們可能還有其它需求，比如說：呼叫插入資料的方法，然後立即從資料庫中查詢這條資料，發現找不到。

如果強行使用 Completer，那麼到最後，可能這個類中會出現一大堆的 Completer ，程式碼難以維護。

解決辦法

其實我們想要的效果是，當有非同步方法在運算元據庫時，別的運算元據的非同步方法應該阻塞住，也就是同一時間只能有一個方法來運算元據庫。我們其實可以使用任務佇列來實現資料庫操作的需求。

我這裡利用 Completer 實現了一個任務佇列:

import 'dart:async';
import 'dart:collection';

/// TaskQueue 不支援 submit await submit, 以下程式碼就存在問題
///
/// TaskQueue taskQueue = TaskQueue();
/// Future<void> task1(String arg)async{
///   await Future.delayed(Duration(milliseconds: 100));
/// }
/// Future<void> task2(String arg)async{
///   在這裡submit時, 任務會被新增到隊尾, 且當前方法任務不會結束
///   新增到隊尾的任務必須等到當前方法任務執行完畢後, 才能繼續執行
///   而隊尾的任務必須等當前任務執行完畢後, 才能執行
///   這就導致相互等待, 使任務無法進行下去
///   解決辦法是, 移除當前的 await, 讓當前任務結束
///   await taskQueue.submit(task1, arg);
/// }
///
/// taskQueue.submit(task2, arg);
///
/// 總結:
/// 被 submit 的方法的內部如果呼叫 submit 方法, 此方法不能 await, 否則任務佇列會被阻塞住
///
/// 如何避免此操作, 可以借鑑以下思想:
/// 以資料庫操作舉例, 有個save方法的邏輯是插入或者更新(先查詢資料庫select,再進行下一步操作);
/// sava方法內部submit,並且select也submit, 就容易出現submit await submit的情況
///
/// 我們可以這樣操作,假設當前類為 DBHelper:
/// 將資料庫的增,刪,查,改操作封裝成私有的 async 方法, 且私有方法不能使用submit
/// DBHelper的公有方法, 可以呼叫自己的私有 async 方法, 但不能呼叫自己的公有方法, 公有方法可以使用submit
/// 這樣就不會存在submit await submit的情況了
class TaskQueue {
  /// 提交任務
  Future<O> submit<A, O>(Function fun, A? arg) async {
    if (!_isEnable) {
      throw Exception('current TaskQueue is recycled.');
    }
    Completer<O> result = new Completer<O>();

    if (!_isStartLoop) {
      _isStartLoop = true;
      _startLoop();
    }

    _queue.addLast(_Runnable<A, O>(
      fun: fun,
      arg: arg,
      completer: result,
    ));
    if (!(_emptyMonitor?.isCompleted ?? true)) {
      _emptyMonitor?.complete();
    }

    return result.future;
  }

  /// 回收 TaskQueue
  void recycle() {
    _isEnable = false;
    if (!(_emptyMonitor?.isCompleted ?? true)) {
      _emptyMonitor?.complete();
    }
    _queue.clear();
  }

  Queue<_Runnable> _queue = Queue<_Runnable>();
  Completer? _emptyMonitor;
  bool _isStartLoop = false;
  bool _isEnable = true;

  Future<void> _startLoop() async {
    while (_isEnable) {
      if (_queue.isEmpty) {
        _emptyMonitor = new Completer();
        await _emptyMonitor!.future;
        _emptyMonitor = null;
      }

      if (!_isEnable) {
        // 當前TaskQueue不可用時, 跳出迴圈
        return;
      }

      _Runnable runnable = _queue.removeFirst();
      try {
        dynamic result = await runnable.fun(runnable.arg);
        runnable.completer.complete(result);
      } catch (e) {
        runnable.completer.completeError(e);
      }
    }
  }
}

class _Runnable<A, O> {
  final Completer<O> completer;
  final Function fun;
  final A? arg;

  _Runnable({
    required this.completer,
    required this.fun,
    this.arg,
  });
}

由於 Flutter 中的 future 不支援暫停操作, 一旦開始執行, 就只能等待執行完。

所以這裡的任務佇列實現是基於方法的延遲呼叫來實現的。

TaskQueue 的用法範例如下:

void main() async {
  Future<void> test1(String data) async {
    await Future.delayed(Duration(milliseconds: 20));
    print('執行了test1');
  }

  Future<String> test2(Map<String, dynamic> args) async {
    await Future.delayed(Duration(milliseconds: 10));
    print('執行了test2');
    return 'mockResult';
  }

  TaskQueue taskQueue = TaskQueue();
  taskQueue.submit(test1, '1');
  taskQueue.submit(test2, {
    'data1': 1,
    'data2': '2',
  }).then((value) {
    print('test2返回結果:${value}');
  });

  await Future.delayed(Duration(milliseconds: 200));
  taskQueue.recycle();
}
/*
執行輸出結果如下:

執行了test1
執行了test2
test2返回結果:mockResult
*/

值得注意的是: 這裡的 TaskQueue 不支援 submit await submit, 原因及範例程式碼已在註釋中說明，這裡不再贅述。

為了避免出現 submit await submit 的情況，我程式碼註釋中也做出了建議（假設當前類為 DBHelper）：

將資料庫的增、刪、查、改操作封裝成私有的非同步方法, 且私有非同步方法不能使用 submit；
DBHelper 的公有方法, 可以呼叫自己的私有非同步方法, 但不能呼叫自己的公有非同步方法, 公有非同步方法可以使用 submit；

這樣就不會出現 submit await submit 的情況了。

於是，上述的資料庫操作範例程式碼就變成了以下的樣子：

class Item {
  int id;
  String data;
  Item({
    required this.id,
    required this.data,
  });
}

class DBTest {
  DBTest._();
  static DBTest instance = DBTest._();
  TaskQueue _taskQueue = TaskQueue();
  bool _existsData = false;
  Future<void> _insert(String data) async {
    // 模擬資料庫插入操作,10毫秒過後,資料庫才有資料
    await Future.delayed(Duration(milliseconds: 10));
    _existsData = true;
    print('執行了插入');
  }

  Future<void> insert(String data) async {
    await _taskQueue.submit(_insert, data);
  }

  Future<void> _update(String data) async {
    // 模擬資料庫更新操作
    await Future.delayed(Duration(milliseconds: 10));
    print('執行了更新');
  }

  Future<void> update(String data) async {
    await _taskQueue.submit(_update, data);
  }

  Future<Item?> _selected(int id) async {
    // 模擬資料庫查詢操作
    await Future.delayed(Duration(milliseconds: 10));
    if (_existsData) {
      // 資料庫中有資料才返回
      return Item(id: 1, data: 'mockData');
    } else {
      // 資料庫沒有資料時,返回null
      return null;
    }
  }

  Future<Item?> selected(int id) async {
    return await _taskQueue.submit(_selected, id);
  }

  /// 先從資料庫表中查詢資料，查詢到了就更新，沒查詢到就插入
  Future<void> _insertOrUpdate(Map<String, dynamic> args) async {
    int id = args['id'];
    String data = args['data'];
    Item? item = await _selected(id);
    if (item == null) {
      await _insert(data);
    } else {
      await _update(data);
    }
  }

  Future<Item?> insertOrUpdate(int id, String data) async {
    return await _taskQueue.submit(_insertOrUpdate, {
      'id': id,
      'data': data,
    });
  }
}

void main() async {
  DBTest.instance.insertOrUpdate(1, 'data');
  DBTest.instance.insertOrUpdate(1, 'data');
}

輸出紀錄檔也變成了我們期望的樣子: