C#（99）：三、.NET 4.0基于任务的异步模式(TAP)，推荐使用

2022-03-20 09:22:08

一、异步编程模型(APM)

二、基于事件的异步编程模式(EAP)

三、基于任务的异步模式(TAP)，推荐使用

四、C# 5.0 新特性——Async和Await使异步编程更简单

一、引言

当使用APM的时候,首先我们要先定义用来包装回调方法的委托,这样难免有点繁琐, 然而使用EAP的时候,我们又需要实现Completed事件和Progress事件，上面两种实现方式感觉都有点繁琐。

同时微软也意识到了这点，所以在.NET 4.0中提出了一个新的异步模式——基于任务的异步模式TAP（Task-based Asynchronous Pattern ）。

基于任务的异步模式 (TAP) 是基于 System.Threading.Tasks.Task 命名空间中的 System.Threading.Tasks.Task<TResult> 和 System.Threading.Tasks 类型，这些类型用于表示任意异步操作。是用于新开发的建议的异步设计模式。

二、什么是TAP——基于任务的异步模式介绍

当看到类中存在TaskAsync为后缀的方法时就代表该类实现了TAP, 并且基于任务的异步模式同样也支持异步操作的取消和进度的报告的功能。

在TAP实现中，我们只需要通过向异步方法传入CancellationToken 参数,因为在异步方法内部会对这个参数的IsCancellationRequested属性进行监控,当异步方法收到一个取消请求时,异步方法将会退出执行。

在TAP中,我们可以通过IProgress<T>接口来实现进度报告的功能。

参考：二、并行编程 - Task任务

1、计算密集型任务

例如，请考虑使用呈现图像的异步方法。

任务的主体可以轮询取消标记，如果在呈现过程中收到取消请求，代码可提前退出。此外，如果启动之前收到取消请求，你需要阻止操作：

internal Task<Bitmap> RenderAsync(ImageData data, CancellationToken cancellationToken)
{
    return Task.Run(() =>
    {
        var bmp = new Bitmap(data.Width, data.Height);
        for(int y=0; y<data.Height; y++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            for(int x=0; x<data.Width; x++)
            {
                // render pixel [x,y] into bmp
            }
        }
        return bmp;
    }, cancellationToken);
}

2、I/O 密集型任务:

假设你想创建一个将在指定时间段后完成的任务。例如，你可能想延迟用户界面中的活动。

System.Threading.Timer 类已提供在指定时间段后以异步方式调用委托的能力，并且你可以通过使用 TaskCompletionSource<TResult> 将 Task<TResult> 前端放在计时器上，例如：

public static Task<DateTimeOffset> Delay(int millisecondsTimeout)
   {
       TaskCompletionSource<DateTimeOffset> tcs = null;
       Timer timer = null;

       timer = new Timer(delegate
       {
           timer.Dispose();
           tcs.TrySetResult(DateTimeOffset.UtcNow);
       }, null, Timeout.Infinite, Timeout.Infinite);

       tcs = new TaskCompletionSource<DateTimeOffset>(timer);
       timer.Change(millisecondsTimeout, Timeout.Infinite);
       return tcs.Task;
   }

从 .NET Framework 4.5 开始，Task.Delay 方法正是为此而提供的，并且你可以在另一个异步方法内使用它。例如，若要实现异步轮询循环：

public static async Task Poll(Uri url, CancellationToken cancellationToken,  IProgress<bool> progress)
{
    while(true)
    {
        await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
        bool success = false;
        try
        {
            await DownloadStringAsync(url);
            success = true;
        }
        catch { /* ignore errors */ }
        progress.Report(success);
    }
}

三、如何使用TAP——使用基于任务的异步模式来异步编程

下面就让我们实现自己的异步方法(亮点为只需要一个方法就可以完成进度报告和异步操作取消的功能)

// Download File CancellationToken 参数赋值获得一个取消请求 progress参数负责进度报告
private void DownLoadFile(string url, CancellationToken ct, IProgress<int> progress)
{
    HttpWebRequest request = null;
    HttpWebResponse response = null;
    Stream responseStream = null;
    int bufferSize = 2048;
    byte[] bufferBytes = new byte[bufferSize];
    try
    {
        request = (HttpWebRequest)WebRequest.Create(url);
        if (DownloadSize != 0)
        {
            request.AddRange(DownloadSize);
        }

        response = (HttpWebResponse)request.GetResponse();
        responseStream = response.GetResponseStream();
        int readSize = 0;
        while (true)
        {
            // 收到取消请求则退出异步操作
            if (ct.IsCancellationRequested == true)
            {
                MessageBox.Show(String.Format("下载暂停，下载的文件地址为：{0}\n 已经下载的字节数为: {1}字节", downloadPath, DownloadSize));

                response.Close();
                filestream.Close();
                sc.Post((state) =>
                {
                    this.btnStart.Enabled = true;
                    this.btnPause.Enabled = false;
                }, null);

                // 退出异步操作
                break;
            }

            readSize = responseStream.Read(bufferBytes, 0, bufferBytes.Length);
            if (readSize > 0)
            {
                DownloadSize += readSize;
                int percentComplete = (int)((float)DownloadSize / (float)totalSize * 100);
                filestream.Write(bufferBytes, 0, readSize);

                // 报告进度
                progress.Report(percentComplete);
            }
            else
            {
                MessageBox.Show(String.Format("下载已完成，下载的文件地址为：{0}，文件的总字节数为: {1}字节", downloadPath, totalSize));

                sc.Post((state) =>
                {
                    this.btnStart.Enabled = false;
                    this.btnPause.Enabled = false;
                }, null);

                response.Close();
                filestream.Close();
                break;
            }
        }
    }
    catch (AggregateException ex)
    {
        // 因为调用Cancel方法会抛出OperationCanceledException异常 将任何OperationCanceledException对象都视为以处理
        ex.Handle(e => e is OperationCanceledException);
    }
}

这样只需要上面的一个方法，我们就可以完成上一专题中文件下载的程序，我们只需要在下载按钮的事件处理程序调用该方法和在暂停按钮的事件处理程序调用CancellationTokenSource.Cancel方法即可，具体代码为：

#region 字段

private int DownloadSize = 0;
private string downloadPath = null;
private long totalSize = 0;
private FileStream filestream;

private CancellationTokenSource cts = null;
private Task task = null;

private SynchronizationContext sc;

#endregion 字段

#region 构造函数

public FileDownLoadForm()
{
    InitializeComponent();
    string url = "http://download.microsoft.com/download/7/0/3/703455ee-a747-4cc8-bd3e-98a615c3aedb/dotNetFx35setup.exe";
    txbUrl.Text = url;
    this.btnPause.Enabled = false;

    // Get Total Size of the download file
    GetTotalSize();
    downloadPath = Environment.GetFolderPath(Environment.SpecialFolder.Desktop) + "\\" + Path.GetFileName(this.txbUrl.Text.Trim());
    if (File.Exists(downloadPath))
    {
        FileInfo fileInfo = new FileInfo(downloadPath);
        DownloadSize = (int)fileInfo.Length;
        if (DownloadSize == totalSize)
        {
            string message = "There is already a file with the same name, do you want to delete it? If not, please change the local path. ";
            var result = MessageBox.Show(message, "File name conflict: " + downloadPath, MessageBoxButtons.OKCancel);

            if (result == System.Windows.Forms.DialogResult.OK)
            {
                File.Delete(downloadPath);
            }
            else
            {
                progressBar1.Value = (int)((float)DownloadSize / (float)totalSize * 100);
                this.btnStart.Enabled = false;
                this.btnPause.Enabled = false;
            }
        }
    }
}

#endregion 构造函数

#region 方法

// Start DownLoad File
private void btnStart_Click(object sender, EventArgs e)
{
    filestream = new FileStream(downloadPath, FileMode.OpenOrCreate);
    this.btnStart.Enabled = false;
    this.btnPause.Enabled = true;

    filestream.Seek(DownloadSize, SeekOrigin.Begin);

    // 捕捉调用线程的同步上下文派生对象
    sc = SynchronizationContext.Current;


    cts = new CancellationTokenSource();
    // 使用指定的操作初始化新的 Task。
    task = new Task(() => Actionmethod(cts.Token), cts.Token);

    // 启动 Task，并将它安排到当前的 TaskScheduler 中执行。
    task.Start();

    //await DownLoadFileAsync(txbUrl.Text.Trim(), cts.Token,new Progress<int>(p => progressBar1.Value = p));
}

// 任务中执行的方法
private void Actionmethod(CancellationToken ct)
{
    // 使用同步上文文的Post方法把更新UI的方法让主线程执行
    DownLoadFile(txbUrl.Text.Trim(), ct, new Progress<int>(p =>
        {
            sc.Post(new SendOrPostCallback((result) => progressBar1.Value = (int)result), p);
        }));
}

// Pause Download
private void btnPause_Click(object sender, EventArgs e)
{
    // 发出一个取消请求
    cts.Cancel();
}

// Get Total Size of File
private void GetTotalSize()
{
    HttpWebRequest myHttpWebRequest = (HttpWebRequest)WebRequest.Create(txbUrl.Text.Trim());
    HttpWebResponse response = (HttpWebResponse)myHttpWebRequest.GetResponse();
    totalSize = response.ContentLength;
    response.Close();
}

四、与其他异步模式和类型互操作

1、从 APM 到 TAP

可以使用 TaskFactory<TResult>.FromAsync 方法来实现此操作的 TAP 包装，如下所示：

public static Task<int> ReadAsync(this Stream stream,  byte[] buffer, int offset,  int count)
{
    if (stream == null) 
       throw new ArgumentNullException("stream");
    
    return Task<int>.Factory.FromAsync(stream.BeginRead,  stream.EndRead, buffer,   offset, count, null);
}

此实现类似于以下内容：

 public static Task<int> ReadAsync(this Stream stream,  byte [] buffer, int offset,   int count)
 {
    if (stream == null) 
        throw new ArgumentNullException("stream");

    var tcs = new TaskCompletionSource<int>();


    stream.BeginRead(buffer, offset, count, iar =>
                     {
                        try { 
                           tcs.TrySetResult(stream.EndRead(iar)); 
                        }
                        catch(OperationCanceledException) { 
                           tcs.TrySetCanceled(); 
                        }
                        catch(Exception exc) { 
                           tcs.TrySetException(exc); 
                        }
                     }, null);
    return tcs.Task;
}

2、从EAP到 TAP

public static Task<string> DownloadStringAsync(Uri url)
 {
     var tcs = new TaskCompletionSource<string>();


     var wc = new WebClient();


     wc.DownloadStringCompleted += (s,e) =>
         {
             if (e.Error != null) 
                tcs.TrySetException(e.Error);
             else if (e.Cancelled) 
                tcs.TrySetCanceled();
             else 
                tcs.TrySetResult(e.Result);
         };
     wc.DownloadStringAsync(url);
     return tcs.Task;
}