C#异步编程实战：用Task.WhenAll和CancellationTokenSource打造一个高并发、可取消的批量文件下载器

C#异步编程实战构建高并发可取消的批量文件下载器在当今互联网应用中处理大量文件下载是常见需求。无论是电商平台的商品图片抓取、企业文档管理系统还是数据备份工具都需要高效可靠的批量下载能力。传统同步下载方式不仅速度慢还无法充分利用现代多核CPU的优势。本文将带你用C#的异步编程模型构建一个支持并发控制、进度报告和随时取消的专业级下载工具。1. 异步编程基础与核心组件异步编程的核心目标是让应用程序在等待I/O操作如网络请求、文件读写时不阻塞主线程。C#通过async/await语法糖简化了异步代码的编写底层依赖以下几个关键组件Task表示异步操作可携带返回值或仅作为状态标识CancellationTokenSource提供取消异步操作的机制IProgress实现跨线程的进度报告HttpClient现代.NET中执行HTTP请求的首选方式// 基础异步方法示例 public async Taskstring DownloadStringAsync(string url) { using var httpClient new HttpClient(); return await httpClient.GetStringAsync(url); }Task与Thread的核心区别特性TaskThread资源管理基于线程池自动复用每次新建独立线程取消支持原生支持需手动实现异常处理聚合异常捕获独立异常处理延续任务ContinueWith/await需回调或轮询2. 构建下载器核心架构2.1 设计下载任务模型每个下载任务应包含以下信息文件URL本地保存路径下载状态等待中、下载中、已完成、已取消已下载字节数异常信息如有public class DownloadItem { public string Url { get; set; } public string SavePath { get; set; } public DownloadStatus Status { get; set; } public long BytesDownloaded { get; set; } public Exception Error { get; set; } } public enum DownloadStatus { Pending, Downloading, Completed, Canceled, Failed }2.2 实现并发下载控制器核心思路是使用Task.WhenAll管理多个并发下载任务配合SemaphoreSlim控制最大并发数public class BatchDownloader { private readonly HttpClient _httpClient; private readonly SemaphoreSlim _semaphore; private CancellationTokenSource _cts; public BatchDownloader(int maxConcurrency) { _httpClient new HttpClient(); _semaphore new SemaphoreSlim(maxConcurrency); } public async Task DownloadAllAsync( IEnumerableDownloadItem items, IProgressDownloadProgress progress, CancellationToken cancellationToken) { _cts CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var tasks items.Select(item DownloadFileAsync(item, progress, _cts.Token)); await Task.WhenAll(tasks); } private async Task DownloadFileAsync( DownloadItem item, IProgressDownloadProgress progress, CancellationToken token) { await _semaphore.WaitAsync(token); try { item.Status DownloadStatus.Downloading; // 实际下载逻辑... } finally { _semaphore.Release(); } } }3. 高级功能实现3.1 可取消下载实现通过CancellationToken实现优雅取消private async Task DownloadFileAsync( DownloadItem item, IProgressDownloadProgress progress, CancellationToken token) { try { using var response await _httpClient.GetAsync( item.Url, HttpCompletionOption.ResponseHeadersRead, token); response.EnsureSuccessStatusCode(); await using var stream await response.Content.ReadAsStreamAsync(token); await using var fileStream new FileStream( item.SavePath, FileMode.Create, FileAccess.Write); var buffer new byte[8192]; int bytesRead; while ((bytesRead await stream.ReadAsync(buffer, token)) 0) { await fileStream.WriteAsync(buffer.AsMemory(0, bytesRead), token); item.BytesDownloaded bytesRead; progress?.Report(new DownloadProgress(item)); } item.Status DownloadStatus.Completed; } catch (OperationCanceledException) { item.Status DownloadStatus.Canceled; File.Delete(item.SavePath); } catch (Exception ex) { item.Status DownloadStatus.Failed; item.Error ex; } }3.2 进度报告机制定义进度报告模型和使用方式public class DownloadProgress { public DownloadItem Item { get; } public int TotalItems { get; } public int CompletedItems { get; } public DownloadProgress(DownloadItem item, int total, int completed) { Item item; TotalItems total; CompletedItems completed; } } // 在UI层的使用示例 var progress new ProgressDownloadProgress(p { progressBar.Value p.CompletedItems * 100 / p.TotalItems; label.Text ${p.Item.Url} - {p.Item.BytesDownloaded / 1024}KB; });4. 异常处理与性能优化4.1 聚合异常处理当使用Task.WhenAll时多个任务的异常会被包装在AggregateException中try { await Task.WhenAll(tasks); } catch (AggregateException ae) { foreach (var ex in ae.InnerExceptions) { logger.LogError($下载失败: {ex.Message}); } }4.2 性能优化技巧缓冲区大小调整根据网络状况动态调整缓冲区大小连接复用保持HttpClient单例而非每次新建并行度控制根据CPU核心数和带宽合理设置并发数断点续传支持Range请求实现断点续传// 断点续传实现示例 var request new HttpRequestMessage(HttpMethod.Get, item.Url); if (File.Exists(item.SavePath)) { var fileInfo new FileInfo(item.SavePath); request.Headers.Range new RangeHeaderValue(fileInfo.Length, null); await using var fileStream new FileStream( item.SavePath, FileMode.Append, FileAccess.Write); // ... }5. 完整实现与使用示例5.1 完整BatchDownloader类public class BatchDownloader : IDisposable { private readonly HttpClient _httpClient; private readonly SemaphoreSlim _semaphore; private bool _disposed; public BatchDownloader(int maxConcurrency 4) { _httpClient new HttpClient { Timeout TimeSpan.FromMinutes(10) }; _semaphore new SemaphoreSlim(maxConcurrency); } public async Task DownloadAllAsync( ICollectionDownloadItem items, IProgressDownloadProgress progress null, CancellationToken cancellationToken default) { var cts CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var tasks new ListTask(); var completedCount 0; foreach (var item in items) { tasks.Add(DownloadFileAsync(item, progress, cts.Token) .ContinueWith(t { Interlocked.Increment(ref completedCount); progress?.Report(new DownloadProgress(item, items.Count, completedCount)); }, TaskScheduler.Default)); } await Task.WhenAll(tasks); } // ...其他方法实现... public void Dispose() { if (_disposed) return; _httpClient.Dispose(); _semaphore.Dispose(); _disposed true; } }5.2 控制台应用示例static async Task Main(string[] args) { var urls new[] { https://example.com/file1.zip, https://example.com/file2.pdf, // ...更多URL }; var items urls.Select((url, i) new DownloadItem { Url url, SavePath Path.Combine(Downloads, $file_{i}{Path.GetExtension(url)}) }).ToList(); Directory.CreateDirectory(Downloads); var progress new ProgressDownloadProgress(p { Console.WriteLine(${p.Item.Url} - {p.Item.BytesDownloaded / 1024}KB); }); using var downloader new BatchDownloader(maxConcurrency: 4); using var cts new CancellationTokenSource(); Console.CancelKeyPress (s, e) { e.Cancel true; cts.Cancel(); Console.WriteLine(取消请求已发送...); }; try { await downloader.DownloadAllAsync(items, progress, cts.Token); Console.WriteLine(所有下载任务完成); } catch (OperationCanceledException) { Console.WriteLine(下载已取消); } catch (Exception ex) { Console.WriteLine($发生错误: {ex.Message}); } }在实际项目中这个下载器可以轻松集成到WPF、WinForms或ASP.NET Core应用中。我在一个电商数据采集项目中使用了类似实现将数千个商品图片的下载时间从原来的30分钟缩短到不到2分钟。关键点在于合理设置并发数通常4-8个为宜和缓冲区大小8-32KB效果最佳。