Flutter 保证数据操作原子性
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');
}
输出日志也变成了我们期望的样子:
执行了插入
执行了更新
总结
Flutter 异步方法修改数据时, 一定要注意数据操作的原子性, 不能因为 Flutter 是单线程架构，就忽略多个异步方法竞争导致数据异常的问题。
Flutter 保证数据操作的原子性，也有可行办法，当逻辑比较简单时，可直接使用 Completer，当逻辑比较复杂时，可以考虑使用任务队列。
另外，本文中的任务队列实现有很大的缺陷，不支持 submit await submit，否则整个任务队列会被阻塞住。
如果诸位有其它的任务队列实现方式，或者保证数据操作原子性的方法，欢迎留言。
