概述

在实际应用中，经常需要同时发起多个网络请求。例如，在一个页面中需要获取用户信息、商品列表、评论数据等。如果依次发起这些请求，总耗时会是各个请求耗时之和；如果并发发起这些请求，总耗时只会是最慢请求的耗时。

合理的并发处理可以显著提高性能，减少用户等待时间。Dart 提供了多种方式来处理并发请求，本文将详细介绍这些方法，并提供完整的代码示例。

为什么需要并发处理

串行请求的问题

串行请求是指依次发起多个请求，前一个请求完成后再发起下一个请求。

请求1 → 请求2 → 请求3 → 请求4
  1秒      2秒      1秒      3秒

总耗时 = 1 + 2 + 1 + 3 = 7秒

并发请求的优势

并发请求是指同时发起多个请求，各个请求并行执行。

请求1 → 请求2 → 请求3 → 请求4
  1秒      2秒      1秒      3秒

总耗时 = max(1, 2, 1, 3) = 3秒

通过并发处理，可以将总耗时从7秒减少到3秒，性能提升超过50%。

使用 Future.wait() 进行并发请求

Future.wait() 是 Dart 中处理并发请求的最常用方法，它会等待所有 Future 完成后返回结果。

基本用法

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

Future<List<dynamic>> fetchMultipleData() async {
  final urls = [
    'https://api.example.com/users',
    'https://api.example.com/posts',
    'https://api.example.com/comments',
  ];

  final futures = urls.map((url) => http.get(Uri.parse(url))).toList();
  final responses = await Future.wait(futures);

  return responses.map((resp) => jsonDecode(resp.body)).toList();
}

代码说明

1. 创建一个 URL 列表
2. 使用 map() 方法将每个 URL 转换为一个 Future
3. 将 Future 列表传递给 Future.wait()
4. Future.wait() 会等待所有 Future 完成，然后返回所有结果

使用示例

try {
  final results = await fetchMultipleData();
  print('用户数据: ${results[0].length}条');
  print('文章数据: ${results[1].length}条');
  print('评论数据: ${results[2].length}条');
} catch (e) {
  print('请求失败: $e');
}

带超时的并发请求

为并发请求设置超时时间，可以避免某个慢请求拖慢整个流程。

实现代码

Future<List<dynamic>> fetchWithTimeout(List<String> urls, {Duration timeout = const Duration(seconds: 10)}) async {
  final futures = urls.map((url) {
    return http.get(Uri.parse(url)).timeout(timeout);
  }).toList();

  try {
    final responses = await Future.wait(futures);
    return responses.map((resp) => jsonDecode(resp.body)).toList();
  } on TimeoutException {
    throw Exception('并发请求超时');
  }
}

代码说明

• 为每个请求单独设置超时时间
• 使用 Future.wait() 等待所有请求完成
• 如果任一请求超时，会抛出 TimeoutException

统一超时 vs 单独超时

// 统一超时：所有请求的总时间不能超过10秒
final responses = await Future.wait(futures).timeout(const Duration(seconds: 10));

// 单独超时：每个请求的时间不能超过5秒
final futures = urls.map((url) {
  return http.get(Uri.parse(url)).timeout(const Duration(seconds: 5));
}).toList();
final responses = await Future.wait(futures);

部分失败的并发请求

默认情况下，Future.wait() 会在任一 Future 失败时立即抛出异常，其他正在执行的请求也会被忽略。但在某些场景下，我们希望即使部分请求失败，也能获取成功请求的结果。

实现代码

Future<List<dynamic?>> fetchWithPartialFailure(List<String> urls) async {
  final futures = urls.map((url) async {
    try {
      final response = await http.get(Uri.parse(url));
      return jsonDecode(response.body);
    } catch (e) {
      print('请求失败: $url, 错误: $e');
      return null;
    }
  }).toList();

  return await Future.wait(futures);
}

代码说明

• 为每个请求单独设置 try-catch
• 如果请求失败，返回 null 而不是抛出异常
• Future.wait() 会等待所有请求完成，包括失败的请求

使用示例

final results = await fetchWithPartialFailure([
  'https://api.example.com/users',
  'https://invalid-url.com/data',
  'https://api.example.com/posts',
]);

// 处理结果
if (results[0] != null) {
  print('用户数据: ${results[0]}');
} else {
  print('用户数据获取失败');
}

if (results[1] != null) {
  print('无效数据: ${results[1]}');
} else {
  print('无效数据获取失败');
}

if (results[2] != null) {
  print('文章数据: ${results[2]}');
} else {
  print('文章数据获取失败');
}

使用 eagerError 参数

Dart 的 Future.wait() 还提供了 eagerError 参数来控制错误处理行为。

// 默认行为：任一失败立即抛出异常
final responses = await Future.wait(futures);

// 等待所有请求完成后再抛出异常
final responses = await Future.wait(futures, eagerError: false);

// 提供清理函数
final responses = await Future.wait(
  futures,
  eagerError: false,
  cleanUp: (successValue) {
    print('清理成功的请求: $successValue');
  },
);

并发请求限制

当需要发起大量请求时，如果同时发起所有请求，可能会给服务器带来过大压力，甚至导致请求被限流。这时需要限制并发请求的数量。

分批处理

Future<List<dynamic>> fetchWithLimit(List<String> urls, {int maxConcurrent = 3}) async {
  List<dynamic> results = [];
  int index = 0;

  while (index < urls.length) {
    int end = (index + maxConcurrent).clamp(0, urls.length);
    List<String> batch = urls.sublist(index, end);

    final futures = batch.map((url) => http.get(Uri.parse(url))).toList();
    final responses = await Future.wait(futures);

    results.addAll(responses.map((resp) => jsonDecode(resp.body)));
    index = end;
  }

  return results;
}

代码说明

1. 初始化结果列表和索引
2. 使用 while 循环分批处理请求
3. 每批最多处理 maxConcurrent 个请求
4. 使用 Future.wait() 并发处理当前批次的请求
5. 将当前批次的结果添加到总结果列表
6. 更新索引，处理下一批请求

使用示例

final urls = List.generate(100, (i) => 'https://api.example.com/data/$i');
final results = await fetchWithLimit(urls, maxConcurrent: 5);
print('获取了 ${results.length} 条数据');

并发限制器类

将并发限制逻辑封装为一个类，可以方便地复用。

class ConcurrentLimiter {
  final int maxConcurrent;

  ConcurrentLimiter({this.maxConcurrent = 3});

  Future<List<T>> execute<T>(
    List<String> urls,
    Future<T> Function(String url) fetcher,
  ) async {
    List<T> results = [];
    int index = 0;

    while (index < urls.length) {
      int end = (index + maxConcurrent).clamp(0, urls.length);
      List<String> batch = urls.sublist(index, end);

      final futures = batch.map((url) => fetcher(url)).toList();
      final responses = await Future.wait(futures);

      results.addAll(responses);
      index = end;
    }

    return results;
  }
}

使用限制器

final limiter = ConcurrentLimiter(maxConcurrent: 5);

final results = await limiter.execute(
  urls,
  (url) async {
    final response = await http.get(Uri.parse(url));
    return jsonDecode(response.body);
  },
);

使用 Stream 处理并发请求

当需要处理大量请求时，可以使用 Stream 来边接收边处理结果，而不需要等待所有请求完成。

实现代码

Stream<dynamic> fetchAsStream(List<String> urls) async* {
  for (var url in urls) {
    try {
      final response = await http.get(Uri.parse(url));
      yield jsonDecode(response.body);
    } catch (e) {
      yield {'error': e.toString()};
    }
  }
}

使用示例

final stream = fetchAsStream(urls);

await for (var result in stream) {
  if (result['error'] != null) {
    print('请求失败: ${result['error']}');
  } else {
    print('获取到数据: $result');
  }
}

并发 Stream

上面的示例是串行处理的，如果需要并发处理并通过 Stream 输出结果，可以使用 Stream.fromFutures()。

Stream<dynamic> fetchConcurrentAsStream(List<String> urls) {
  final futures = urls.map((url) async {
    try {
      final response = await http.get(Uri.parse(url));
      return jsonDecode(response.body);
    } catch (e) {
      return {'error': e.toString()};
    }
  }).toList();

  return Stream.fromFutures(futures);
}

使用示例

final stream = fetchConcurrentAsStream(urls);

stream.listen((result) {
  if (result['error'] != null) {
    print('请求失败: ${result['error']}');
  } else {
    print('获取到数据: $result');
  }
}, onDone: () {
  print('所有请求完成');
});

高级并发模式

1. Future.wait2() 和 Future.wait3()

Dart 提供了专门处理2个或3个 Future 的方法，可以获得更好的类型安全。

// 处理2个 Future
final (user, posts) = await Future.wait2(
  http.get(Uri.parse('https://api.example.com/users/1')),
  http.get(Uri.parse('https://api.example.com/posts')),
);

// 处理3个 Future  
final (user, posts, comments) = await Future.wait3(
  http.get(Uri.parse('https://api.example.com/users/1')),
  http.get(Uri.parse('https://api.example.com/posts')),
  http.get(Uri.parse('https://api.example.com/comments')),
);

2. Future.any()

Future.any() 返回第一个完成的 Future 的结果，其他 Future 会被忽略。

final fastest = await Future.any([
  http.get(Uri.parse('https://api1.example.com/data')),
  http.get(Uri.parse('https://api2.example.com/data')),
  http.get(Uri.parse('https://api3.example.com/data')),
]);

print('最快的响应: ${fastest.statusCode}');

3. Future.doWhile()

Future.doWhile() 重复执行一个异步操作，直到返回 false。

int count = 0;
await Future.doWhile(() async {
  print('执行第 ${++count} 次');
  await http.get(Uri.parse('https://api.example.com/data'));
  return count < 5;
});

UI 层的并发请求处理

在 UI 层需要展示并发请求的进度和结果。

示例代码

class ConcurrentDemoPage extends StatefulWidget {
  const ConcurrentDemoPage({super.key});

  @override
  State<ConcurrentDemoPage> createState() => _ConcurrentDemoPageState();
}

class _ConcurrentDemoPageState extends State<ConcurrentDemoPage> {
  List<String> _results = [];
  bool _loading = false;
  int _completedCount = 0;

  Future<void> _fetchData() async {
    setState(() {
      _loading = true;
      _results = [];
      _completedCount = 0;
    });

    final urls = [
      'https://jsonplaceholder.typicode.com/users/1',
      'https://jsonplaceholder.typicode.com/users/2',
      'https://jsonplaceholder.typicode.com/users/3',
      'https://jsonplaceholder.typicode.com/users/4',
    ];

    final futures = urls.map((url) async {
      try {
        final response = await http.get(Uri.parse(url));
        var data = jsonDecode(response.body);
        
        setState(() {
          _completedCount++;
          _results.add('用户${data['id']}: ${data['name']}');
        });

        return data;
      } catch (e) {
        setState(() {
          _completedCount++;
          _results.add('请求失败: $e');
        });
        return null;
      }
    }).toList();

    await Future.wait(futures);

    setState(() => _loading = false);
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('并发请求示例')),
      body: Padding(
        padding: const EdgeInsets.all(16),
        child: Column(
          children: [
            ElevatedButton(
              onPressed: _loading ? null : _fetchData,
              child: _loading 
                  ? const CircularProgressIndicator() 
                  : const Text('发起并发请求'),
            ),
            const SizedBox(height: 20),
            if (_loading)
              Text('进度: $_completedCount/4'),
            const SizedBox(height: 20),
            Expanded(
              child: ListView.builder(
                itemCount: _results.length,
                itemBuilder: (context, index) {
                  return Card(
                    child: Padding(
                      padding: const EdgeInsets.all(8),
                      child: Text(_results[index]),
                    ),
                  );
                },
              ),
            ),
          ],
        ),
      ),
    );
  }
}

并发请求最佳实践

1. 使用 Future.wait() 同时发起多个请求

对于相互独立的请求，使用 Future.wait() 可以显著提高性能。

2. 设置合理的超时时间

为每个请求设置合理的超时时间，避免慢请求拖慢整个流程。

3. 处理部分失败的情况

在需要容错的场景下，为每个请求单独处理错误，确保部分失败不影响其他请求。

4. 限制并发请求数量

当需要发起大量请求时，限制并发数量可以避免服务器压力过大。

5. 使用 Stream 处理大量请求

对于大量请求，可以使用 Stream 边接收边处理，减少内存占用。

6. 考虑使用 Isolate

对于 CPU 密集型任务，可以考虑使用 Isolate 来避免阻塞主线程。

7. 监控并发请求

监控并发请求的性能和成功率，及时发现和解决问题。

并发模式对比

模式	特点	适用场景
Future.wait()	等待所有请求完成，任一失败则全部失败	所有请求都必须成功的场景
Future.wait(eagerError: false)	等待所有请求完成，收集成功和失败	需要容错的场景
Future.any()	返回第一个完成的请求	多个数据源取最快的场景
Stream	边接收边处理	大量请求或需要实时展示的场景
分批处理	限制并发数量	大量请求需要控制并发的场景

总结

请求并发处理是提高网络请求性能的关键。通过合理使用 Future.wait()、Stream 和分批处理等方式，可以显著减少用户等待时间，提升用户体验。

在实际开发中，需要根据具体场景选择合适的并发模式，并注意处理错误和限制并发数量，以构建健壮的网络请求系统。

完整示例代码

以下是一个完整的并发请求处理示例：

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

class ConcurrentApiClient {
  final String baseUrl;
  final int maxConcurrent;

  ConcurrentApiClient({
    required this.baseUrl,
    this.maxConcurrent = 3,
  });

  Future<List<dynamic>> fetchMultiple(List<String> paths) async {
    final urls = paths.map((path) => '$baseUrl$path').toList();
    final futures = urls.map((url) => http.get(Uri.parse(url))).toList();
    final responses = await Future.wait(futures);
    return responses.map((resp) => jsonDecode(resp.body)).toList();
  }

  Future<List<dynamic>> fetchWithLimit(List<String> paths) async {
    List<dynamic> results = [];
    int index = 0;
    final urls = paths.map((path) => '$baseUrl$path').toList();

    while (index < urls.length) {
      int end = (index + maxConcurrent).clamp(0, urls.length);
      List<String> batch = urls.sublist(index, end);

      final futures = batch.map((url) => http.get(Uri.parse(url))).toList();
      final responses = await Future.wait(futures);

      results.addAll(responses.map((resp) => jsonDecode(resp.body)));
      index = end;
    }

    return results;
  }

  Future<List<dynamic?>> fetchWithPartialFailure(List<String> paths) async {
    final urls = paths.map((path) => '$baseUrl$path').toList();
    final futures = urls.map((url) async {
      try {
        final response = await http.get(Uri.parse(url));
        return jsonDecode(response.body);
      } catch (e) {
        return null;
      }
    }).toList();
    return await Future.wait(futures);
  }
}

通过这个示例，你可以快速实现一个支持多种并发模式的 API 客户端，为你的网络请求提供高效的并发处理能力。