MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

利用 Future 构建 Flutter 应用的网络请求队列

2023-05-121.2k 阅读

1. Flutter 网络请求基础与 Future 简介

在 Flutter 应用开发中，网络请求是非常常见的操作。无论是获取服务器数据展示在界面上，还是向服务器提交用户数据，都离不开网络请求。Flutter 提供了多种方式来进行网络请求，其中最常用的是 http 包。

1.1 http 包基本使用

使用 http 包进行简单的 GET 请求示例如下：

import 'package:http/http.dart' as http;

Future<String> fetchData() async {
  final response = await http.get(Uri.parse('https://example.com/api/data'));
  if (response.statusCode == 200) {
    return response.body;
  } else {
    throw Exception('Failed to load data');
  }
}

在上述代码中，通过 http.get 方法发起一个 GET 请求到指定的 URL。await 关键字用于等待异步操作（即网络请求）完成，并返回响应结果。如果响应状态码为 200，则返回响应体；否则抛出异常。

1.2 Future 的概念

Future 是 Dart 中用于处理异步操作的核心类。它表示一个可能在未来某个时刻完成的操作，并最终返回一个结果。Future 可以处于三种状态：未完成（pending）、已完成（completed）且有值、已完成且有错误。

例如，当发起一个网络请求时，这个请求不会立即完成，而是返回一个 Future 对象。我们可以使用 await 关键字来暂停当前代码的执行，直到 Future 完成，然后获取其结果。

Future<int> delayedResult() {
  return Future.delayed(const Duration(seconds: 2), () => 42);
}

void main() async {
  print('Start');
  int result = await delayedResult();
  print('Result: $result');
  print('End');
}

在这个例子中，delayedResult 函数返回一个 Future，它会在延迟 2 秒后返回值 42。await 使得 print('Result: $result') 这行代码等待 delayedResult 完成后才执行，输出结果为：

Start
// 等待 2 秒
Result: 42
End

Future 还提供了一系列方法来处理异步操作的结果，如 then、catchError 等。例如，上述代码也可以写成：

Future<int> delayedResult() {
  return Future.delayed(const Duration(seconds: 2), () => 42);
}

void main() {
  print('Start');
  delayedResult()
    .then((result) => print('Result: $result'))
    .catchError((error) => print('Error: $error'))
    .whenComplete(() => print('End'));
}

这种方式通过链式调用处理 Future 的结果，then 用于处理成功的结果，catchError 用于捕获错误，whenComplete 无论成功或失败都会执行。

2. 为什么需要网络请求队列

在实际的 Flutter 应用中，可能会遇到多个网络请求同时发起的情况。例如，一个复杂的界面可能需要从多个 API 端点获取数据来填充不同的部分。然而，过多的并发网络请求可能会带来一些问题。

2.1 资源消耗问题

每个网络请求都需要占用一定的系统资源，如网络带宽、内存等。如果同时发起大量的网络请求，可能会耗尽系统资源，导致应用性能下降，甚至出现卡顿或崩溃的情况。特别是在移动设备上，资源相对有限，这种情况更加明显。

2.2 网络稳定性影响

过多的并发请求可能会使网络连接变得不稳定。在网络状况不佳时，过多的请求可能导致部分请求超时或失败。而且，某些服务器可能对单个客户端的并发请求数量有限制，如果超过这个限制，服务器可能会拒绝请求。

2.3 数据一致性问题

在一些情况下，网络请求之间可能存在依赖关系。例如，请求 B 需要依赖请求 A 的结果。如果并发执行这些请求，可能会导致数据不一致的问题，因为请求 B 可能在请求 A 完成之前就开始执行了。

为了解决这些问题，引入网络请求队列是一种有效的方法。网络请求队列可以控制请求的发送顺序和并发数量，确保应用在网络请求方面更加稳定和高效。

3. 利用 Future 构建网络请求队列

3.1 基本思路

利用 Future 构建网络请求队列的基本思路是将所有的网络请求封装成 Future，然后按照一定的顺序依次执行这些 Future。我们可以使用一个 List 来存储这些 Future，并通过 Future 的相关方法来控制它们的执行。

3.2 简单队列实现

下面是一个简单的网络请求队列实现示例：

import 'package:http/http.dart' as http;

class RequestQueue {
  final List<Future> _requests = [];

  void addRequest(Future request) {
    _requests.add(request);
  }

  Future executeQueue() {
    return Future(() => null)
      .then((_) => _requests.fold(Future(() => null), (prev, next) {
            return prev.then((_) => next);
          }));
  }
}

void main() async {
  RequestQueue queue = RequestQueue();

  Future<String> request1() async {
    final response = await http.get(Uri.parse('https://example.com/api/request1'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 1 failed');
    }
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  queue.addRequest(request1());
  queue.addRequest(request2());

  try {
    await queue.executeQueue();
    print('All requests completed successfully');
  } catch (error) {
    print('An error occurred: $error');
  }
}

在上述代码中，RequestQueue 类用于管理网络请求队列。addRequest 方法用于将请求添加到队列中，executeQueue 方法用于执行队列中的所有请求。

executeQueue 方法通过 Future(() => null) 开始一个新的 Future，然后使用 fold 方法对队列中的 Future 进行遍历。fold 方法会依次执行每个 Future，前一个 Future 完成后才会执行下一个，从而实现请求的顺序执行。

3.3 并发控制的队列实现

有时候，我们可能不希望所有请求都顺序执行，而是希望在一定的并发数量内执行。下面是一个支持并发控制的网络请求队列实现：

import 'package:http/http.dart' as http;
import 'dart:async';

class RequestQueue {
  final int _concurrency;
  final List<Future> _requests = [];
  final List<Future> _activeRequests = [];

  RequestQueue(this._concurrency);

  void addRequest(Future request) {
    _requests.add(request);
  }

  Future executeQueue() {
    return Future(() => null).then((_) {
      _startNextRequests();
      return Future.wait(_activeRequests);
    });
  }

  void _startNextRequests() {
    while (_activeRequests.length < _concurrency && _requests.isNotEmpty) {
      Future request = _requests.removeAt(0);
      _activeRequests.add(request.then((_) {
        _activeRequests.remove(request);
        _startNextRequests();
      }).catchError((error) {
        _activeRequests.remove(request);
        throw error;
      }));
    }
  }
}

void main() async {
  RequestQueue queue = RequestQueue(2);

  Future<String> request1() async {
    final response = await http.get(Uri.parse('https://example.com/api/request1'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 1 failed');
    }
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  Future<String> request3() async {
    final response = await http.get(Uri.parse('https://example.com/api/request3'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 3 failed');
    }
  }

  queue.addRequest(request1());
  queue.addRequest(request2());
  queue.addRequest(request3());

  try {
    await queue.executeQueue();
    print('All requests completed successfully');
  } catch (error) {
    print('An error occurred: $error');
  }
}

在这个实现中，RequestQueue 类的构造函数接受一个 _concurrency 参数，表示允许同时执行的请求数量。_activeRequests 列表用于存储当前正在执行的请求。

executeQueue 方法首先开始一个空的 Future，然后调用 _startNextRequests 方法开始执行请求。_startNextRequests 方法会在 _activeRequests 数量小于 _concurrency 且 _requests 不为空时，从 _requests 中取出一个请求并添加到 _activeRequests 中执行。当一个请求完成（无论是成功还是失败），它会从 _activeRequests 中移除，并再次调用 _startNextRequests 方法，以启动下一个请求。

4. 实际应用场景中的优化与注意事项

4.1 错误处理优化

在实际应用中，错误处理是非常重要的。当一个网络请求失败时，我们可能不希望整个队列立即停止执行，而是希望能够继续执行其他请求，并在最后统一处理错误。

import 'package:http/http.dart' as http;
import 'dart:async';

class RequestQueue {
  final int _concurrency;
  final List<Future> _requests = [];
  final List<dynamic> _results = [];
  final List<dynamic> _errors = [];

  RequestQueue(this._concurrency);

  void addRequest(Future request) {
    _requests.add(request);
  }

  Future executeQueue() {
    return Future(() => null).then((_) {
      _startNextRequests();
      return Future.wait(_activeRequests, eagerError: false).then((results) {
        _results.addAll(results);
        return _results;
      }).catchError((error) {
        _errors.add(error);
        return _results;
      });
    });
  }

  void _startNextRequests() {
    while (_activeRequests.length < _concurrency && _requests.isNotEmpty) {
      Future request = _requests.removeAt(0);
      _activeRequests.add(request.then((result) {
        _activeRequests.remove(request);
        _startNextRequests();
        return result;
      }).catchError((error) {
        _activeRequests.remove(request);
        _errors.add(error);
        _startNextRequests();
        return null;
      }));
    }
  }
}

void main() async {
  RequestQueue queue = RequestQueue(2);

  Future<String> request1() async {
    final response = await http.get(Uri.parse('https://example.com/api/request1'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 1 failed');
    }
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  Future<String> request3() async {
    final response = await http.get(Uri.parse('https://example.com/api/request3'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 3 failed');
    }
  }

  queue.addRequest(request1());
  queue.addRequest(request2());
  queue.addRequest(request3());

  List<dynamic> results = await queue.executeQueue();
  print('Results: $results');
  print('Errors: $_errors');
}

在这个优化版本中，_results 列表用于存储成功请求的结果，_errors 列表用于存储失败请求的错误信息。Future.wait 的 eagerError 参数设置为 false，这样即使有请求失败，其他请求也会继续执行。在每个请求的 catchError 中，将错误信息添加到 _errors 列表中，并继续执行下一个请求。

4.2 请求优先级处理

在一些应用场景中，某些网络请求可能具有更高的优先级。例如，用户登录请求可能比获取广告数据的请求更重要，需要优先执行。

我们可以通过给请求添加优先级标识，并在队列执行逻辑中根据优先级进行处理。以下是一个简单的实现示例：

import 'package:http/http.dart' as http;
import 'dart:async';

class Request {
  final Future future;
  final int priority;

  Request(this.future, this.priority);
}

class RequestQueue {
  final int _concurrency;
  final List<Request> _requests = [];
  final List<Future> _activeRequests = [];

  RequestQueue(this._concurrency);

  void addRequest(Request request) {
    _requests.add(request);
  }

  Future executeQueue() {
    return Future(() => null).then((_) {
      _startNextRequests();
      return Future.wait(_activeRequests);
    });
  }

  void _startNextRequests() {
    _requests.sort((a, b) => b.priority - a.priority);
    while (_activeRequests.length < _concurrency && _requests.isNotEmpty) {
      Request request = _requests.removeAt(0);
      _activeRequests.add(request.future.then((_) {
        _activeRequests.remove(request.future);
        _startNextRequests();
      }).catchError((error) {
        _activeRequests.remove(request.future);
        throw error;
      }));
    }
  }
}

void main() async {
  RequestQueue queue = RequestQueue(2);

  Future<String> request1() async {
    final response = await http.get(Uri.parse('https://example.com/api/request1'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 1 failed');
    }
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  Future<String> request3() async {
    final response = await http.get(Uri.parse('https://example.com/api/request3'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 3 failed');
    }
  }

  queue.addRequest(Request(request1(), 2));
  queue.addRequest(Request(request2(), 1));
  queue.addRequest(Request(request3(), 3));

  try {
    await queue.executeQueue();
    print('All requests completed successfully');
  } catch (error) {
    print('An error occurred: $error');
  }
}

在这个示例中，Request 类包含了请求的 Future 和优先级 priority。在 _startNextRequests 方法中，首先对 _requests 列表按照优先级进行排序，然后再依次执行请求，从而实现了优先级处理。

4.3 缓存处理

为了提高应用性能，减少不必要的网络请求，可以引入缓存机制。对于一些不经常变化的数据请求，可以将其结果缓存起来，下次请求相同数据时直接从缓存中获取。

import 'package:http/http.dart' as http;
import 'dart:async';
import 'dart:collection';

class RequestQueue {
  final int _concurrency;
  final List<Future> _requests = [];
  final List<Future> _activeRequests = [];
  final Map<String, dynamic> _cache = {};

  RequestQueue(this._concurrency);

  void addRequest(Future request, String cacheKey) {
    if (_cache.containsKey(cacheKey)) {
      _requests.add(Future.value(_cache[cacheKey]));
    } else {
      _requests.add(request.then((result) {
        _cache[cacheKey] = result;
        return result;
      }));
    }
  }

  Future executeQueue() {
    return Future(() => null).then((_) {
      _startNextRequests();
      return Future.wait(_activeRequests);
    });
  }

  void _startNextRequests() {
    while (_activeRequests.length < _concurrency && _requests.isNotEmpty) {
      Future request = _requests.removeAt(0);
      _activeRequests.add(request.then((_) {
        _activeRequests.remove(request);
        _startNextRequests();
      }).catchError((error) {
        _activeRequests.remove(request);
        throw error;
      }));
    }
  }
}

void main() async {
  RequestQueue queue = RequestQueue(2);

  Future<String> request1() async {
    final response = await http.get(Uri.parse('https://example.com/api/request1'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 1 failed');
    }
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  queue.addRequest(request1(), 'request1_cache_key');
  queue.addRequest(request2(), 'request2_cache_key');
  queue.addRequest(request1(), 'request1_cache_key');

  try {
    await queue.executeQueue();
    print('All requests completed successfully');
  } catch (error) {
    print('An error occurred: $error');
  }
}

在上述代码中，_cache 是一个 Map，用于存储缓存数据。addRequest 方法在添加请求时，会先检查缓存中是否存在对应的数据。如果存在，则直接添加一个已完成的 Future 到队列中；否则，添加请求 Future，并在请求成功后将结果存入缓存。

4.4 内存管理与资源释放

在处理大量网络请求时，需要注意内存管理和资源释放。例如，当一个请求不再需要时，应该及时取消相关的资源占用，如网络连接、定时器等。

对于 http 包的网络请求，可以通过 HttpClient 的 close 方法来关闭连接。在 Future 中，可以使用 Completer 来手动控制 Future 的完成，并在适当的时候取消相关操作。

import 'package:http/http.dart' as http;
import 'dart:async';
import 'dart:io';

class RequestQueue {
  final int _concurrency;
  final List<Future> _requests = [];
  final List<Future> _activeRequests = [];

  RequestQueue(this._concurrency);

  void addRequest(Future request) {
    _requests.add(request);
  }

  Future executeQueue() {
    return Future(() => null).then((_) {
      _startNextRequests();
      return Future.wait(_activeRequests);
    });
  }

  void _startNextRequests() {
    while (_activeRequests.length < _concurrency && _requests.isNotEmpty) {
      Future request = _requests.removeAt(0);
      _activeRequests.add(request.then((_) {
        _activeRequests.remove(request);
        _startNextRequests();
      }).catchError((error) {
        _activeRequests.remove(request);
        if (error is SocketException) {
          // 处理网络连接相关错误，关闭连接
          (error.osError as SocketException).socket?.close();
        }
        throw error;
      }));
    }
  }
}

void main() async {
  RequestQueue queue = RequestQueue(2);

  Future<String> request1() async {
    Completer<String> completer = Completer<String>();
    Timer(Duration(seconds: 5), () {
      if (!completer.isCompleted) {
        completer.completeError('Request timed out');
      }
    });
    try {
      final response = await http.get(Uri.parse('https://example.com/api/request1'));
      if (response.statusCode == 200) {
        completer.complete(response.body);
      } else {
        completer.completeError('Request 1 failed');
      }
    } catch (error) {
      completer.completeError(error);
    }
    return completer.future;
  }

  Future<String> request2() async {
    final response = await http.get(Uri.parse('https://example.com/api/request2'));
    if (response.statusCode == 200) {
      return response.body;
    } else {
      throw Exception('Request 2 failed');
    }
  }

  queue.addRequest(request1());
  queue.addRequest(request2());

  try {
    await queue.executeQueue();
    print('All requests completed successfully');
  } catch (error) {
    print('An error occurred: $error');
  }
}

在这个示例中，对于 request1，使用 Completer 手动控制 Future 的完成，并设置了一个 5 秒的定时器。如果请求在 5 秒内未完成，则通过 completer.completeError 来完成 Future 并抛出超时错误。在 _startNextRequests 方法的 catchError 中，对于 SocketException 类型的错误，关闭相关的 Socket 连接，以释放资源。

通过以上的优化和注意事项处理，可以使基于 Future 构建的 Flutter 应用网络请求队列更加健壮、高效，满足实际应用场景的需求。无论是在复杂的企业级应用还是小型的移动应用中，合理管理网络请求队列都是提升应用性能和用户体验的重要环节。