8天玩轉並行開發——第六天 異步編程模型

     

      在.net里面異步編程模型由來已久，相信大家也知道Begin/End異步模式和事件異步模式，在task出現以后，這些東西都可以被task包裝

起來，可能有人會問，這樣做有什么好處，下面一一道來。

 

一： Begin/End模式

1： 委托

    在執行委托方法的時候，我們常常會看到一個Invoke，同時也有一對你或許不常使用的BeginInvoke，EndInvoke方法對，當然Invoke方法

是阻塞主線程，而BeginInvoke則是另開一個線程。

    class Program
    {
        static void Main(string[] args)
        {
            var func = new Func<string, string>(i => { return i + "i can fly"; });

            var state = func.BeginInvoke("yes,", Callback, func);

            Console.Read();
        }

        static void Callback(IAsyncResult async)
        {
            var result = async.AsyncState as Func<string, string>;

            Console.WriteLine(result.EndInvoke(async));
        }
    }

 

下面我們用task包裝一下

    class Program
    {
        static void Main(string[] args)
        {
            var func = new Func<string, string>(i =>
            {
                return i + "i can fly";
            });

            Task<string>.Factory.FromAsync(func.BeginInvoke, func.EndInvoke, "yes,", null).ContinueWith
                (i =>
                {
                    Console.WriteLine(i.Result);
                });

            Console.Read();
        }
    }

 

可以看出，task只要一句就搞定，體現了task的第一個優點：簡潔。

 

2：流

    我們發現在Stream抽象類中提供了這樣兩對BeginRead/EndRead,BeginWrite/EndWrite（異步讀寫）的方法，這樣它的n多繼承類都可以

實現異步讀寫，下面舉個繼承類FileStream的例子。

 static void Main(string[] args)
        {
            var path = "C://1.txt";

            FileStream fs = new FileStream(path, FileMode.Open);

            FileInfo info = new FileInfo(path);

            byte[] b = new byte[info.Length];

            var asycState = fs.BeginRead(b, 0, b.Length, (result) =>
            {
                var file = result.AsyncState as FileStream;

                Console.WriteLine("文件內容：{0}", Encoding.Default.GetString(b));

                file.Close();

            }, fs);

            Console.WriteLine("我是主線程，我不會被阻塞！");

            Console.Read();
        }

 

我們用task包裝一下

   static void Main(string[] args)
        {
            var path = "C://1.txt";

            FileStream fs = new FileStream(path, FileMode.Open);

            FileInfo info = new FileInfo(path);

            byte[] b = new byte[info.Length];

            Task<int>.Factory.FromAsync(fs.BeginRead, fs.EndRead, b, 0, b.Length, null, TaskCreationOptions.None)
                .ContinueWith
                (i =>
                {
                    Console.WriteLine("文件內容：{0}", Encoding.Default.GetString(b));
                });

            Console.WriteLine("我是主線程，我不會被阻塞！");

            Console.Read();
        }

 

其實看到這里，我們並沒有發現task還有其他的什么優點，但是深入的想一下其實並不是這么回事，task能夠游刃於線程並發和同步，而原始的異步

編程要實現線程同步還是比較麻煩的。

 

     假如現在有這樣的一個需求，我們需要從3個txt文件中讀取字符，然后進行倒序，前提是不能阻塞主線程。如果不用task的話我可能會用工作線程

去監視一個bool變量來判斷文件是否全部讀取完畢，然后再進行倒序，我也說了，相對task來說還是比較麻煩的，這里我就用task來實現。

    class Program
    {
        static byte[] b;

        static void Main()
        {
            string[] array = { "C://1.txt", "C://2.txt", "C://3.txt" };

            List<Task<string>> taskList = new List<Task<string>>(3);

            foreach (var item in array)
            {
                taskList.Add(ReadAsyc(item));
            }

            Task.Factory.ContinueWhenAll(taskList.ToArray(), i =>
            {
                string result = string.Empty;

                //獲取各個task返回的結果
                foreach (var item in i)
                {
                    result += item.Result;
                }

                //倒序
                String content = new String(result.OrderByDescending(j => j).ToArray());

                Console.WriteLine("倒序結果："+content);
            });

            Console.WriteLine("我是主線程，我不會被阻塞");

            Console.ReadKey();
        }

        //異步讀取
        static Task<string> ReadAsyc(string path)
        {
            FileInfo info = new FileInfo(path);

            byte[] b = new byte[info.Length];

            FileStream fs = new FileStream(path, FileMode.Open);

            Task<int> task = Task<int>.Factory.FromAsync(fs.BeginRead, fs.EndRead, b, 0, b.Length, null, TaskCreationOptions.None);

            //返回當前task的執行結果
            return task.ContinueWith(i =>
            {
                return i.Result > 0 ? Encoding.Default.GetString(b) : string.Empty;
            }, TaskContinuationOptions.ExecuteSynchronously);
        }
    }

 

可以看出，task的第二個優點就是：靈活性。

 

這里可能就有人要問了，能不能用開多個線程用read以同步的形式讀取，變相的實現文件異步讀取，或許我們可能常聽說程序優化后，最后出現的

瓶頸在IO上面，是的，IO是比較耗費資源的，要命的是如果我們開的是工作線程走IO讀取文件，那么該線程就會一直處於等待狀態，不會再接收任

何的外來請求，直到線程讀取到文件為止，那么我們能不能用更少的線程來應對更多的IO操作呢？答案肯定是可以的，這里就設計到了”異步IO“的

概念，具體內容可以參照百科：http://baike.baidu.com/view/1865389.htm  ，有幸的是beginXXX,endXXX完美的封裝了“異步IO”。

 

二：事件模式

   這個模式常以XXXCompleted的形式結尾，我們在文件下載這一塊會經常遇到，這里我也舉個例子。

    class Program
    {
        static void Main(string[] args)
        {
            WebClient client = new WebClient();

            client.DownloadFileCompleted += new System.ComponentModel.AsyncCompletedEventHandler(client_DownloadFileCompleted);

            client.DownloadFileAsync(new Uri("http://imgsrc.baidu.com/baike/abpic/item/6a600c338744ebf844a0bc74d9f9d72a6159a7ac.jpg"),
                                   "1.jpg", "圖片下完了，你懂的！");

            Console.WriteLine("我是主線程，我不會被阻塞！");
            Console.Read();
        }

        static void client_DownloadFileCompleted(object sender, System.ComponentModel.AsyncCompletedEventArgs e)
        {
            Console.WriteLine("\n" + e.UserState);
        }
    }

 

先前也說了，task是非常靈活的，那么針對這種異步模型，我們該如何封裝成task來使用，幸好framework中提供了TaskCompletionSource來幫助

我們快速實現。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading.Tasks;
using System.Net;
using System.ComponentModel;

namespace ConsoleApplication4
{
    class Program
    {
        static void Main()
        {
            var downloadTask = DownLoadFileInTask(
                    new Uri(@"http://www.7720mm.cn/uploadfile/2010/1120/20101120073035736.jpg")
                    , "C://1.jpg");

            downloadTask.ContinueWith(i =>
            {
                Console.WriteLine("圖片:" + i.Result + "下載完畢！");
            });

            Console.WriteLine("我是主線程，我不會被阻塞!");

            Console.Read();
        }

        static Task<string> DownLoadFileInTask(Uri address, string saveFile)
        {
            var wc = new WebClient();

            var tcs = new TaskCompletionSource<string>(address);

            //處理異步操作的一個委托
            AsyncCompletedEventHandler handler = null;

            handler = (sender, e) =>
            {
                if (e.Error != null)
                {
                    tcs.TrySetException(e.Error);
                }
                else
                {
                    if (e.Cancelled)
                    {
                        tcs.TrySetCanceled();
                    }
                    else
                    {
                        tcs.TrySetResult(saveFile);
                    }
                }

                wc.DownloadFileCompleted -= handler;
            };

            //我們將下載事件與我們自定義的handler進行了關聯
            wc.DownloadFileCompleted += handler;

            try
            {
                wc.DownloadFileAsync(address, saveFile);
            }
            catch (Exception ex)
            {
                wc.DownloadFileCompleted -= handler;

                tcs.TrySetException(ex);
            }

            return tcs.Task;
        }
    }
}