C#綜合揭密－細說多執行緒（下）

引言

本文主要從執行緒的基礎用法，CLR執行緒池當中工作者執行緒與I/O執行緒的開發，並行操作PLINQ等多個面向介紹多執行緒的開發。
其中委託的BeginInvoke方法以及回呼函數最為常用。
而 I/O執行緒可能容易遭到大家的忽略，其實在開發多執行緒系統，更應該多留意I/O執行緒的操作。特別是在ASP.NET開發當中，可能更多人只會留意在客戶端使用Ajax或是在伺服器端使用UpdatePanel。其實合理使用I/O線程在通訊項目或文件下載時，能盡量降低IIS的壓力。
平行程式設計是Framework4.0中極力推廣的非同步操作方式，更值得更深入學習。
希望這篇文章能對各位的學習研究有所幫助，當中有所錯漏的地方敬請點評。

五、CLR線程池的I/O線程

在前一節所介紹的線程都屬於CLR線程池的工作者線程，這一節開始為大家介紹一下CLR線程池的I/O線程

I/O 線程是.NET專為存取外部資源所設定的一種線程，因為訪問外部資源常常要受到外界因素的影響，為了防止讓主線程受影響而長期處於阻塞狀態，.NET為多個I /O操作都建立起了非同步方法，例如：FileStream、TCP/IP、WebRequest、WebService等等，而且每個非同步方法的使用方式都非常類似，都是以BeginXXX為開始，以EndXXX結束，下面為大家一一解說。

 

5.1  非同步讀取寫入 FileStream

需要在 FileStream 非同步呼叫 I/O線程，必須使用下列建構子建立 FileStream 對象，並將useAsync設為 true。

FileStream stream = new FileStream ( string path, FileMode mode, FileAccess access, FileShare share, int bufferSize,bool useAsync ) ;

其中 path 是檔案的相對路徑或絕對路徑或絕對路徑或絕對路徑決定如何建立檔案 存取檔案的方式; share 決定檔案如何進程共享; bufferSize 是代表緩衝區大小,一般預設最小值為8，在啟動非同步讀取或寫入時，檔案大小一般大於緩衝大小; userAsync代表是否啟動非同步I/ O線程。

注意：當使用BeginRead 和BeginWrite 方法在執行大量讀取或寫入時效果更好，但對於少量的讀/寫，這些方法速度可能比同步讀取還要慢，因為進行線程間的切換需要大量時間。

 

5.1.1 非同步寫入

FileStream中包含BeginWrite、EndWrite 方法可以啟動I/O執行緒進行非同步寫入。

public override IAsyncResult BeginWrite ( byte[] array, int offset, int numBytes, AsyncCallback userCallback, Object stateObject )
public override v問題syncResult, 使用方式與委託的BeginInvoke方法相似，最好就是使用回呼函數，避免執行緒阻塞。在最後兩個參數中，參數AsyncCallback用於綁定回呼函數; 參數Object用於傳遞外部資料。要注意一點：AsyncCallback所綁定的回呼函數必須是單一 IAsyncResult 參數的無回傳值方法。

在例子中，把FileStream當作外部資料傳遞到回呼函數當中，然後在回呼函數中利用IAsyncResult.AsyncState取得FileStream對象，最後透過FileStream.EndWrite（IAsyncResult）結束寫入。

class Program
    {
        static void Main(string[] args)
        {
            //把线程池的最大值设置为1000
            ThreadPool.SetMaxThreads(1000, 1000);
            ThreadPoolMessage("Start");

            //新立文件File.sour
            FileStream stream = new FileStream("File.sour", FileMode.OpenOrCreate, 
                                       FileAccess.ReadWrite,FileShare.ReadWrite,1024,true);
            byte[] bytes = new byte[16384];
            string message = "An operating-system ThreadId has no fixed relationship........";
            bytes = Encoding.Unicode.GetBytes(message);

            //启动异步写入
            stream.BeginWrite(bytes, 0, (int)bytes.Length,new AsyncCallback(Callback),stream);
            stream.Flush();
            
            Console.ReadKey();
        }

        static void Callback(IAsyncResult result)
        {
            //显示线程池现状
            Thread.Sleep(200);
            ThreadPoolMessage("AsyncCallback");
            //结束异步写入
            FileStream stream = (FileStream)result.AsyncState;
            stream.EndWrite(result);
            stream.Close();
        }

        //显示线程池现状
        static void ThreadPoolMessage(string data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
                  "WorkerThreads is:{2}  CompletionPortThreads is :{3}",
                  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

由輸出結果可以看到，在使用FileStream.BeginWrite方法後，系統會自動啟動CLR執行緒池中I/O執行緒。

5.1.2 非同步讀取

FileStream 中包含 BeginRead 與 EndRead 可以非同步呼叫I/O執行緒進行讀取。

public override IAsyncResult BeginRead ( byte[] array,int offset,int numBytes, AsyncCallback userCallback,Object stateObject)

public override int AsyncCallback用來綁定回調函數; Object用於傳遞外部資料。在回呼函數只需要使用IAsyncResut.AsyncState就可取得外部資料。 EndWrite 方法會傳回從流讀取到的位元組數量。

先定義 FileData 類，裡麵包含FileStream對象，byte[] 陣列和長度。然後把FileData物件當作外部資料傳到回呼函數，在回呼函數中，把IAsyncResult.AsyncState強制轉換成FileData，然後透過FileStream.EndRead（IAsyncResult）結束讀取。最後比較一下長度，若讀取到的長度與輸入的資料長度不一至，則拋出例外。

class Program
     {
         public class FileData
         {
             public FileStream Stream;
             public int Length;
             public byte[] ByteData;
         }
 
         static void Main(string[] args)
         {       
             //把线程池的最大值设置为1000
             ThreadPool.SetMaxThreads(1000, 1000);
             ThreadPoolMessage("Start");
             ReadFile();
 
             Console.ReadKey();
         }
 
         static void ReadFile()
         {
             byte[] byteData=new byte[80961024];
             FileStream stream = new FileStream("File1.sour", FileMode.OpenOrCreate, 
                                     FileAccess.ReadWrite, FileShare.ReadWrite, 1024, true);
             
             //把FileStream对象,byte[]对象，长度等有关数据绑定到FileData对象中，以附带属性方式送到回调函数
             FileData fileData = new FileData();
             fileData.Stream = stream;
             fileData.Length = (int)stream.Length;
             fileData.ByteData = byteData;
             
             //启动异步读取
             stream.BeginRead(byteData, 0, fileData.Length, new AsyncCallback(Completed), fileData);
         }
  
         static void Completed(IAsyncResult result)
         {
             ThreadPoolMessage("Completed");
 
             //把AsyncResult.AsyncState转换为FileData对象，以FileStream.EndRead完成异步读取
             FileData fileData = (FileData)result.AsyncState;
             int length=fileData.Stream.EndRead(result);
             fileData.Stream.Close();
 
             //如果读取到的长度与输入长度不一致，则抛出异常
             if (length != fileData.Length)
                 throw new Exception("Stream is not complete!");
 
             string data=Encoding.ASCII.GetString(fileData.ByteData, 0, fileData.Length);
             Console.WriteLine(data.Substring(2,22));
         }
 
         //显示线程池现状
         static void ThreadPoolMessage(string data)
         {
             int a, b;
             ThreadPool.GetAvailableThreads(out a, out b);
             string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
                          "WorkerThreads is:{2}  CompletionPortThreads is :{3}",
                          data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
             Console.WriteLine(message);      
         }
             
   }

由輸出結果可以看到，在使用FileStream.BeginRead方法後，系統會自動啟動CLR執行緒池中I/O執行緒。

注意：如果你看到的测试结果正好相反：工作者线程为999，I/O线程为1000，这是因为FileStream的文件容量小于缓冲值1024所致的。此时文件将会一次性读取或写入，而系统将启动工作者线程而非I/O线程来处理回调函数。

5.2 异步操作TCP/IP套接字

在介绍 TCP/IP 套接字前先简单介绍一下 NetworkStream 类，它是用于网络访问的基础数据流。 NetworkStream 提供了好几个方法控制套接字数据的发送与接收, 其中BeginRead、EndRead、BeginWrite、EndWrite 能够实现异步操作，而且异步线程是来自于CLR线程池的I/O线程。

public override int ReadByte ()
public override int Read (byte[] buffer,int offset, int size)

public override void WriteByte (byte value)
public override void Write (byte[] buffer,int offset, int size)

public override IAsyncResult BeginRead (byte [] buffer, int offset, int size, AsyncCallback callback, Object state )
public override int EndRead(IAsyncResult result)

public override IAsyncResult BeginWrite (byte [] buffer, int offset, int size, AsyncCallback callback, Object state )
public override void EndWrite(IAsyncResult result)

若要创建 NetworkStream，必须提供已连接的 Socket。而在.NET中使用TCP/IP套接字不需要直接与Socket打交道，因为.NET把Socket的大部分操作都放在System.Net.TcpListener和System.Net.Sockets.TcpClient里面，这两个类大大地简化了Socket的操作。一般套接字对象Socket包含一个Accept（）方法，此方法能产生阻塞来等待客户端的请求，而在TcpListener类里也包含了一个相似的方法 public TcpClient AcceptTcpClient（）用于等待客户端的请求。此方法将会返回一个TcpClient 对象，通过 TcpClient 的 public NetworkStream GetStream（）方法就能获取NetworkStream对象，控制套接字数据的发送与接收。

下面以一个例子说明异步调用TCP/IP套接字收发数据的过程。

首先在服务器端建立默认地址127.0.0.1用于收发信息，使用此地址与端口500新建TcpListener对象，调用TcpListener.Start 侦听传入的连接请求，再使用一个死循环来监听信息。

在ChatClient类包括有接收信息与发送信息两个功能：当接收到客户端请求时，它会利用 NetworkStream.BeginRead 读取客户端信息，并在回调函数ReceiveAsyncCallback中输出信息内容，若接收到的信息的大小小于1时，它将会抛出一个异常。当信息成功接收后，再使用 NetworkStream.BeginWrite 方法回馈信息到客户端

class Program
    {
        static void Main(string[] args)
        {
            //设置CLR线程池最大线程数
            ThreadPool.SetMaxThreads(1000, 1000);
   
            //默认地址为127.0.0.1
            IPAddress ipAddress = IPAddress.Parse("127.0.0.1");
            TcpListener tcpListener = new TcpListener(ipAddress, 500);
            tcpListener.Start();
            
            //以一个死循环来实现监听
            while (true)
            {   //调用一个ChatClient对象来实现监听
                ChatClient chatClient = new ChatClient(tcpListener.AcceptTcpClient());    
            }
        }
    }

    public class ChatClient
    {
        static TcpClient tcpClient;
        static byte[] byteMessage;
        static string clientEndPoint;

        public ChatClient(TcpClient tcpClient1)
        {
            tcpClient = tcpClient1;
            byteMessage = new byte[tcpClient.ReceiveBufferSize];
           
            //显示客户端信息
            clientEndPoint = tcpClient.Client.RemoteEndPoint.ToString();
            Console.WriteLine("Client's endpoint is " + clientEndPoint);
            
            //使用NetworkStream.BeginRead异步读取信息
            NetworkStream networkStream = tcpClient.GetStream();
            networkStream.BeginRead(byteMessage, 0, tcpClient.ReceiveBufferSize,
                                         new AsyncCallback(ReceiveAsyncCallback), null);
        }

        public void ReceiveAsyncCallback(IAsyncResult iAsyncResult)
        {
            //显示CLR线程池状态
            Thread.Sleep(100);
            ThreadPoolMessage("\nMessage is receiving");

            //使用NetworkStream.EndRead结束异步读取
            NetworkStream networkStreamRead = tcpClient.GetStream();
            int length=networkStreamRead.EndRead(iAsyncResult);

            //如果接收到的数据长度少于1则抛出异常
            if (length < 1)
            {
                tcpClient.GetStream().Close();
                throw new Exception("Disconnection!");
            }

            //显示接收信息
            string message = Encoding.UTF8.GetString(byteMessage, 0, length);
            Console.WriteLine("Message:" + message);

            //使用NetworkStream.BeginWrite异步发送信息
            byte[] sendMessage = Encoding.UTF8.GetBytes("Message is received!");
            NetworkStream networkStreamWrite=tcpClient.GetStream();
            networkStreamWrite.BeginWrite(sendMessage, 0, sendMessage.Length, 
                                            new AsyncCallback(SendAsyncCallback), null);
        }

        //把信息转换成二进制数据，然后发送到客户端
        public void SendAsyncCallback(IAsyncResult iAsyncResult)
        {
            //显示CLR线程池状态
            Thread.Sleep(100);
            ThreadPoolMessage("\nMessage is sending");

            //使用NetworkStream.EndWrite结束异步发送
            tcpClient.GetStream().EndWrite(iAsyncResult);

            //重新监听
            tcpClient.GetStream().BeginRead(byteMessage, 0, tcpClient.ReceiveBufferSize,
                                               new AsyncCallback(ReceiveAsyncCallback), null);
        }

        //显示线程池现状
        static void ThreadPoolMessage(string data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n  CurrentThreadId is {1}\n  " +
                  "WorkerThreads is:{2}  CompletionPortThreads is :{3}\n",
                  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }
    }

而在客户端只是使用简单的开发方式，利用TcpClient连接到服务器端，然后调用NetworkStream.Write方法发送信息，最后调用NetworkStream.Read方法读取回馈信息

static void Main(string[] args)
        {
            //连接服务端
            TcpClient tcpClient = new TcpClient("127.0.0.1", 500);

            //发送信息
            NetworkStream networkStream = tcpClient.GetStream();
            byte[] sendMessage = Encoding.UTF8.GetBytes("Client request connection!");
            networkStream.Write(sendMessage, 0, sendMessage.Length);
            networkStream.Flush();

            //接收信息
            byte[] receiveMessage=new byte[1024];
            int count=networkStream.Read(receiveMessage, 0,1024);
            Console.WriteLine(Encoding.UTF8.GetString(receiveMessage));
            Console.ReadKey();
        }

注意观察运行结果，服务器端的异步操作线程都是来自于CLR线程池的I/O线程

5.3 异步WebRequest

System.Net.WebRequest 是 .NET 为实现访问 Internet 的 “请求/响应模型” 而开发的一个 abstract 基类， 它主要有三个子类：FtpWebRequest、HttpWebRequest、FileWebRequest。当使用WebRequest.Create（string uri）创建对象时，应用程序就可以根据请求协议判断实现类来进行操作。FileWebRequest、FtpWebRequest、HttpWebRequest 各有其作用：FileWebRequest 使用 “file://路径” 的URI方式实现对本地资源和内部文件的请求/响应、FtpWebRequest 使用FTP文件传输协议实现文件请求/响应、HttpWebRequest 用于处理HTTP的页面请求/响应。由于使用方法相类似，下面就以常用的HttpWebRequest为例子介绍一下异步WebRequest的使用方法。

在使用ASP.NET开发网站的时候，往往会忽略了HttpWebRequest的使用，因为开发都假设客户端是使用浏览器等工具去阅读页面的。但如果你对REST开发方式有所了解，那对 HttpWebRequest 就应该非常熟悉。它可以在路径参数、头文件、页面主体、Cookie 等多处地方加入请求条件，然后对回复数据进行适当处理。HttpWebRequest 包含有以下几个常用方法用于处理请求/响应：

public override Stream GetRequestStream ()
public override WebResponse GetResponse ()

public override IAsyncResult BeginGetRequestStream ( AsyncCallback callback, Object state )
public override Stream EndGetRequestStream ( IAsyncResult asyncResult )
public override IAsyncResult BeginGetResponse ( AsyncCallback callback, Object state )
public override WebResponse EndGetResponse ( IAsyncResult asyncResult )

其中BeginGetRequestStream、EndGetRequestStream 用于异步向HttpWebRequest对象写入请求信息; BeginGetResponse、EndGetResponse 用于异步发送页面请求并获取返回信息。使用异步方式操作Internet的“请求/响应”，避免主线程长期处于等待状态，而操作期间异步线程是来自CLR线程池的I/O线程。

注意：请求与响应不能使用同步与异步混合开发模式，即当请求写入使用GetRequestStream同步模式，即使响应使用BeginGetResponse异步方法，操作也与GetRequestStream方法在于同一线程内。

下面以简单的例子介绍一下异步请求的用法。

首先为Person类加上可序列化特性，在服务器端建立Hanlder.ashx，通过Request.InputStream 获取到请求数据并把数据转化为String对象，此实例中数据是以 “Id：1” 的形式实现传送的。然后根据Id查找对应的Person对象，并把Person对象写入Response.OutStream 中返还到客户端。

在客户端先把 HttpWebRequird.Method 设置为 "post"，使用异步方式通过BeginGetRequireStream获取请求数据流，然后写入请求数据 “Id:1”。再使用异步方法BeginGetResponse 获取回复数据，最后把数据反序列化为Person对象显示出来。

注意：HttpWebRequire.Method默认为get，在写入请求前必须把HttpWebRequire.Method设置为post,否则在使用BeginGetRequireStream 获取请求数据流的时候，系统就会发出 “无法发送具有此谓词类型的内容正文" 的异常。

Model

namespace Model
{
    [Serializable]
    public class Person
    {
        public int ID
        {
            get;
            set;
        }
        public string Name
        {
            get;
            set;
        }
        public int Age
        {
            get;
            set;
        }
    }
}

服务器端

public class Handler : IHttpHandler {

    public void ProcessRequest(HttpContext context)
    {
        //把信息转换为String，找出输入条件Id
        byte[] bytes=new byte[1024];
        int length=context.Request.InputStream.Read(bytes,0,1024);
        string condition = Encoding.Default.GetString(bytes);
        int id = int.Parse(condition.Split(new string[] { ":" }, 
                           StringSplitOptions.RemoveEmptyEntries)[1]);
        
        //根据Id查找对应Person对象
        var person = GetPersonList().Where(x => x.ID == id).First();
        
        //所Person格式化为二进制数据写入OutputStream
        BinaryFormatter formatter = new BinaryFormatter();
        formatter.Serialize(context.Response.OutputStream, person);
    }

    //模拟源数据
    private IList<Person> GetPersonList()
    {
        var personList = new List<Person>();
        
        var person1 = new Person();
        person1.ID = 1;
        person1.Name = "Leslie";
        person1.Age = 30;
        personList.Add(person1);
        ...........
        return personList;
    }

    public bool IsReusable
    {
        get { return true;}
    }
}

客户端

class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(1000, 1000);
            Request();
            Console.ReadKey();
        }

        static void Request()
        {
            ThreadPoolMessage("Start"); 
            //使用WebRequest.Create方法建立HttpWebRequest对象
            HttpWebRequest webRequest = (HttpWebRequest)WebRequest.Create(
                                            "http://localhost:5700/Handler.ashx");
            webRequest.Method = "post";
           
            //对写入数据的RequestStream对象进行异步请求
            IAsyncResult result=webRequest.BeginGetRequestStream(
                new AsyncCallback(EndGetRequestStream),webRequest);
        }

        static void EndGetRequestStream(IAsyncResult result)
        {
            ThreadPoolMessage("RequestStream Complete");
            //获取RequestStream
            HttpWebRequest webRequest = (HttpWebRequest)result.AsyncState;
            Stream stream=webRequest.EndGetRequestStream(result);

            //写入请求条件
            byte[] condition = Encoding.Default.GetBytes("Id:1");
            stream.Write(condition, 0, condition.Length);

            //异步接收回传信息
            IAsyncResult responseResult = webRequest.BeginGetResponse(
                new AsyncCallback(EndGetResponse), webRequest);
        }

        static void EndGetResponse(IAsyncResult result)
        {
            //显出线程池现状
            ThreadPoolMessage("GetResponse Complete");

            //结束异步请求，获取结果
            HttpWebRequest webRequest = (HttpWebRequest)result.AsyncState;
            WebResponse webResponse = webRequest.EndGetResponse(result);
            
            //把输出结果转化为Person对象
            Stream stream = webResponse.GetResponseStream();
            BinaryFormatter formatter = new BinaryFormatter();
            var person=(Person)formatter.Deserialize(stream);
            Console.WriteLine(string.Format("Person    Id:{0} Name:{1} Age:{2}",
                person.ID, person.Name, person.Age));
        }

        //显示线程池现状
        static void ThreadPoolMessage(string data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n  CurrentThreadId is {1}\n  " +
                  "WorkerThreads is:{2}  CompletionPortThreads is :{3}\n",
                  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }
    }

从运行结果可以看到，BeginGetRequireStream、BeginGetResponse方法是使用CLR线程池的I/O线程。

5.4 异步调用WebService

相比TCP/IP套接字，在使用WebService的时候，服务器端需要更复杂的操作处理，使用时间往往会更长。为了避免客户端长期处于等待状态，在配置服务引用时选择 “生成异步操作”，系统可以自动建立异步调用的方式。

以.NET 2.0以前，系统都是使用ASMX来设计WebService，而近年来WCF可说是火热登场，下面就以WCF为例子简单介绍一下异步调用WebService的例子。

由于系统可以自动生成异步方法，使用起来非常简单，首先在服务器端建立服务ExampleService，里面包含方法Method。客户端引用此服务时，选择 “生成异步操作”。然后使用 BeginMethod 启动异步方法， 在回调函数中调用EndMethod结束异步调用。

服务端

[ServiceContract]
     public interface IExampleService
     {
         [OperationContract]
         string Method(string name);
     }
 
     public class ExampleService : IExampleService
     {
         public string Method(string name)
         {
             return "Hello " + name;
         }
     }
 
     class Program
     {
         static void Main(string[] args)
         {
             ServiceHost host = new ServiceHost(typeof(ExampleService));
             host.Open();
             Console.ReadKey();
             host.Close();
          }
     }
 
 <configuration>
     <system.serviceModel>
         <services>
             <service name="Example.ExampleService">
                 <endpoint address="" binding="wsHttpBinding" contract="Example.IExampleService">
                     <identity>
                         <dns value="localhost" />
                     </identity>
                 </endpoint>
                 <endpoint address="mex" binding="mexHttpBinding" contract="IMetadataExchange" />
                 <host>
                     <baseAddresses>
                         <add baseAddress="http://localhost:7200/Example/ExampleService/" />
                     </baseAddresses>
                 </host>
             </service>
         </services>
     </system.serviceModel>
 </configuration>

客户端

class Program
     {
         static void Main(string[] args)
         {
             //设置最大线程数
             ThreadPool.SetMaxThreads(1000, 1000);
             ThreadPoolMessage("Start");
             
             //建立服务对象，异步调用服务方法
             ExampleServiceReference.ExampleServiceClient exampleService = new
                                     ExampleServiceReference.ExampleServiceClient();
             exampleService.BeginMethod("Leslie",new AsyncCallback(AsyncCallbackMethod), 
                                         exampleService);  
             Console.ReadKey();
         }
 
         static void AsyncCallbackMethod(IAsyncResult result)
         {
             Thread.Sleep(1000);
             ThreadPoolMessage("Complete");
             ExampleServiceReference.ExampleServiceClient example =
                 (ExampleServiceReference.ExampleServiceClient)result.AsyncState;
             string data=example.EndMethod(result);
             Console.WriteLine(data);
         }
 
         //显示线程池现状
         static void ThreadPoolMessage(string data)
         {
             int a, b;
             ThreadPool.GetAvailableThreads(out a, out b);
             string message = string.Format("{0}\n  CurrentThreadId is {1}\n  " +
                   "WorkerThreads is:{2}  CompletionPortThreads is :{3}\n",
                   data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
 
             Console.WriteLine(message);
         }
     }
 
 <configuration>
     <system.serviceModel>
         <bindings>
             <wsHttpBinding>
                 <binding name="WSHttpBinding_IExampleService" closeTimeout="00:01:00"
                     openTimeout="00:01:00" receiveTimeout="00:10:00" sendTimeout="00:01:00"
                     bypassProxyOnLocal="false" transactionFlow="false" 
                     hostNameComparisonMode="StrongWildcard" maxBufferPoolSize="524288"
                     maxReceivedMessageSize="65536" messageEncoding="Text" textEncoding="utf-8"
                     useDefaultWebProxy="true" allowCookies="false">
                     <readerQuotas maxDepth="32" maxStringContentLength="8192" maxArrayLength="16384"
                         maxBytesPerRead="4096" maxNameTableCharCount="16384" />
                     <reliableSession ordered="true" inactivityTimeout="00:10:00" enabled="false" />
                     <security mode="Message">
                         <transport clientCredentialType="Windows" proxyCredentialType="None"
                           realm="" />
                         <message clientCredentialType="Windows" negotiateServiceCredential="true"
                             algorithmSuite="Default" />
                     </security>
                 </binding>
             </wsHttpBinding>
         </bindings>
         <client>
             <endpoint address="http://localhost:7200/Example/ExampleService/"
                 binding="wsHttpBinding" bindingConfiguration="WSHttpBinding_IExampleService"
                 contract="ExampleServiceReference.IExampleService" 
                 name="WSHttpBinding_IExampleService">
                 <identity>
                     <dns value="localhost" />
                 </identity>
             </endpoint>
         </client>
     </system.serviceModel>
 </configuration>

注意观察运行结果，异步调用服务时，回调函数都是运行于CLR线程池的I/O线程当中。

六、异步 SqlCommand

从ADO.NET 2.0开始，SqlCommand就新增了几个异步方法执行SQL命令。相对于同步执行方式，它使主线程不需要等待数据库的返回结果，在使用复杂性查询或批量插入时将有效提高主线程的效率。使用异步SqlCommand的时候，请注意把ConnectionString 的 Asynchronous Processing 设置为 true 。

注意：SqlCommand异步操作的特别之处在于线程并不依赖于CLR线程池，而是由Windows内部提供，这比使用异步委托更有效率。但如果需要使用回调函数的时候，回调函数的线程依然是来自于CLR线程池的工作者线程。

SqlCommand有以下几个方法支持异步操作：

public IAsyncResult BeginExecuteNonQuery (......)
public int EndExecuteNonQuery(IAsyncResult)

public IAsyncResult BeginExecuteReader(......)
public SqlDataReader EndExecuteReader(IAsyncResult)

public IAsyncResult BeginExecuteXmlReader (......)
public XmlReader EndExecuteXmlReader(IAsyncResult）

由于使用方式相似，此处就以 BeginExecuteNonQuery 为例子，介绍一下异步SqlCommand的使用。首先建立connectionString,注意把Asynchronous Processing设置为true来启动异步命令，然后把SqlCommand.CommandText设置为 WAITFOR DELAY "0:0:3" 来虚拟数据库操作。再通过BeginExecuteNonQuery启动异步操作，利用轮询方式监测操作情况。最后在操作完成后使用EndExecuteNonQuery完成异步操作。

class Program
    {
        //把Asynchronous Processing设置为true
        static string connectionString = "Data Source=LESLIE-PC;Initial Catalog=Business;“+
                                         "Integrated Security=True;Asynchronous Processing=true";

        static void Main(string[] args)
        {
            //把CLR线程池最大线程数设置为1000
            ThreadPool.SetMaxThreads(1000, 1000);
            ThreadPoolMessage("Start");

            //使用WAITFOR DELAY命令来虚拟操作
            SqlConnection connection = new SqlConnection(connectionString);
            SqlCommand command = new SqlCommand("WAITFOR DELAY '0:0:3';", connection);
            connection.Open();

            //启动异步SqlCommand操作，利用轮询方式监测操作
            IAsyncResult result = command.BeginExecuteNonQuery();
            ThreadPoolMessage("BeginRead");
            while (!result.AsyncWaitHandle.WaitOne(500))
                Console.WriteLine("Main thread do work........");

            //结束异步SqlCommand
            int count= command.EndExecuteNonQuery(result);
            ThreadPoolMessage("\nCompleted");
            Console.ReadKey();
        }

        //显示线程池现状
        static void ThreadPoolMessage(string data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
                   "WorkerThreads is:{2}  CompletionPortThreads is :{3}\n",
                   data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

注意运行结果，SqlCommand的异步执行线程并不属于CLR线程池。

如果觉得使用轮询方式过于麻烦，可以使用回调函数，但要注意当调用回调函数时，线程是来自于CLR线程池的工作者线程。

class Program
    {
        //把Asynchronous Processing设置为true
        static string connectionString = "Data Source=LESLIE-PC;Initial Catalog=Business;”+
                                         “Integrated Security=True;Asynchronous Processing=true";
        static void Main(string[] args)
        {
            //把CLR线程池最大线程数设置为1000
            ThreadPool.SetMaxThreads(1000, 1000);
            ThreadPoolMessage("Start");

            //使用WAITFOR DELAY命令来虚拟操作
            SqlConnection connection = new SqlConnection(connectionString);
            SqlCommand command = new SqlCommand("WAITFOR DELAY '0:0:3';", connection);
            connection.Open();
            
            //启动异步SqlCommand操作，并把SqlCommand对象传递到回调函数
            IAsyncResult result = command.BeginExecuteNonQuery(
                                       new AsyncCallback(AsyncCallbackMethod),command);
            Console.ReadKey();
        }

        static void AsyncCallbackMethod(IAsyncResult result)
        {
            Thread.Sleep(200);
            ThreadPoolMessage("AsyncCallback");
            SqlCommand command = (SqlCommand)result.AsyncState;
            int count=command.EndExecuteNonQuery(result);
            command.Connection.Close();
        }

        //显示线程池现状
        static void ThreadPoolMessage(string data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
                  "WorkerThreads is:{2}  CompletionPortThreads is :{3}\n",
                  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }
    }

运行结果：

七、并行编程与PLINQ

要使用多线程开发，必须非常熟悉Thread的使用，而且在开发过程中可能会面对很多未知的问题。为了简化开发，.NET 4.0 特别提供一个并行编程库System.Threading.Tasks，它可以简化并行开发，你无需直接跟线程或线程池打交道，就可以简单建立多线程应用程序。此外，.NET还提供了新的一组扩展方法PLINQ，它具有自动分析查询功能，如果并行查询能提高系统效率，则同时运行，如果查询未能从并行查询中受益，则按原顺序查询。下面将详细介绍并行操作的方式。

7.1 泛型委托

使用并行编程可以同时操作多个委托,在介绍并行编程前先简单介绍一下两个泛型委托System.Func<>与System.Action<>。

Func<>是一个能接受多个参数和一个返回值的泛型委托，它能接受0个到16个输入参数, 其中 T1,T2,T3,T4......T16 代表自定的输入类型，TResult为自定义的返回值。
public delegate TResult Func（）
public delegate TResult Func（T1 arg1）
public delegate TResult Func（T1 arg1,T2 arg2）
public delegate TResult Func（T1 arg1,T2 arg2,T3 arg3）
public delegate TResult Func（T1 arg1,T2 arg2,T3 arg3,T4 arg4）
..............
public delegate TResult Func（T1 arg1,T2 arg2,T3 arg3,T4 arg4，...... ,T16 arg16）

Action<>与Func<>十分相似，不同在于Action<>的返回值为void，Action能接受0~16个参数
public delegate void Action（）
public delegate void Action（T1 arg1,T2 arg2）
public delegate void Action（T1 arg1,T2 arg2, T3 arg3）
.............
public delegate void Action（T1 arg1,T2 arg2,T3 arg3,T4 arg4，...... ,T16 arg16）

7.2 任务并行库（TPL）

System.Threading.Tasks中的类被统称为任务并行库（Task Parallel Library，TPL），TPL使用CLR线程池把工作分配到CPU，并能自动处理工作分区、线程调度、取消支持、状态管理以及其他低级别的细节操作，极大地简化了多线程的开发。

注意：TPL比Thread更具智能性，当它判断任务集并没有从并行运行中受益，就会选择按顺序运行。但并非所有的项目都适合使用并行开发，创建过多并行任务可能会损害程序的性能，降低运行效率。

TPL包括常用的数据并行与任务并行两种执行方式：

7.2.1 数据并行

数据并行的核心类就是System.Threading.Tasks.Parallel，它包含两个静态方法 Parallel.For 与 Parallel.ForEach, 使用方式与for、foreach相仿。通过这两个方法可以并行处理System.Func<>、System.Action<>委托。

以下一个例子就是利用 public static ParallelLoopResult For( int from, int max, Action) 方法对List进行并行查询。
假设使用单线程方式查询3个Person对象，需要用时大约6秒，在使用并行方式，只需使用2秒就能完成查询，而且能够避开Thread的繁琐处理。

class Program
    {
        static void Main(string[] args)
        {
            //设置最大线程数
            ThreadPool.SetMaxThreads(1000, 1000);
            //并行查询
            Parallel.For(0, 3,n =>
                {
                    Thread.Sleep(2000);  //模拟查询
                    ThreadPoolMessage(GetPersonList()[n]);
                });
            Console.ReadKey();
        }

        //模拟源数据
        static IList<Person> GetPersonList()
        {
            var personList = new List<Person>();

            var person1 = new Person();
            person1.ID = 1;
            person1.Name = "Leslie";
            person1.Age = 30;
            personList.Add(person1);
            ...........
            return personList;
        }

        //显示线程池现状
        static void ThreadPoolMessage(Person person)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
                  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
                  "  CompletionPortThreads is :{5}\n",
                  person.ID, person.Name, person.Age,
                  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }
    }

观察运行结果，对象并非按照原排列顺序进行查询，而是使用并行方式查询。

若想停止操作，可以利用ParallelLoopState参数，下面以ForEach作为例子。
public static ParallelLoopResult ForEach( IEnumerable source, Action action)
其中source为数据集，在Action委托的ParallelLoopState参数当中包含有Break（）和 Stop（）两个方法都可以使迭代停止。Break的使用跟传统for里面的使用方式相似，但因为处于并行处理当中，使用Break并不能保证所有运行能立即停止，在当前迭代之前的迭代会继续执行。若想立即停止操作，可以使用Stop方法，它能保证立即终止所有的操作，无论它们是处于当前迭代的之前还是之后。

class Program
    {
         static void Main(string[] args)
         {
             //设置最大线程数
             ThreadPool.SetMaxThreads(1000, 1000);
 
             //并行查询
             Parallel.ForEach(GetPersonList(), (person, state) =>
                 {
                     if (person.ID == 2)
                         state.Stop();
                     ThreadPoolMessage(person);
                 });
             Console.ReadKey();
         }
 
         //模拟源数据
         static IList<Person> GetPersonList()
         {
             var personList = new List<Person>();
 
             var person1 = new Person();
             person1.ID = 1;
             person1.Name = "Leslie";
             person1.Age = 30;
             personList.Add(person1);
             ..........
             return personList;
         }
 
         //显示线程池现状
         static void ThreadPoolMessage(Person person)
         {
             int a, b;
             ThreadPool.GetAvailableThreads(out a, out b);
             string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
                   "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
                   "  CompletionPortThreads is :{5}\n",
                   person.ID, person.Name, person.Age,
                   Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
 
             Console.WriteLine(message);
         }
     }

观察运行结果，当Person的ID等于2时，运行将会停止。

当要在多个线程中调用本地变量，可以使用以下方法：
public static ParallelLoopResult ForEach(IEnumerable, Func, Func, Action)
其中第一个参数为数据集;
第二个参数是一个Func委托，用于在每个线程执行前进行初始化;
第 三个参数是委托Func,它能对数据集的每个成员进行迭代，当中T1是数据集的成员，T2是一个ParallelLoopState对 象，它可以控制迭代的状态，T3是线程中的本地变量;
第四个参数是一个Action委托，用于对每个线程的最终状态进行最终操作。

在以下例子中，使用ForEach计算多个Order的总体价格。在ForEach方法中，首先把参数初始化为0f，然后用把同一个Order的多个OrderItem价格进行累加，计算出Order的价格，最后把多个Order的价格进行累加，计算出多个Order的总体价格。

public class Order
    {
        public int ID;
        public float Price;
    }

    public class OrderItem
    {
        public int ID;
        public string Goods;
        public int OrderID;
        public float Price;
        public int Count;
    }

    class Program
    {
        static void Main(string[] args)
        {
            //设置最大线程数
            ThreadPool.SetMaxThreads(1000, 1000);
            float totalPrice = 0f;
            //并行查询
            var parallelResult = Parallel.ForEach(GetOrderList(),
                     () => 0f,   //把参数初始值设为0
                     (order, state, orderPrice) =>
                     {
                         //计算单个Order的价格
                         orderPrice = GetOrderItem().Where(item => item.OrderID == order.ID)
                              .Sum(item => item.Price * item.Count);
                         order.Price = orderPrice;
                         ThreadPoolMessage(order);
                         
                         return orderPrice;
                     },
                    (finallyPrice) =>
                    {
                        totalPrice += finallyPrice;//计算多个Order的总体价格
                    }
                );
            
            while (!parallelResult.IsCompleted)
                Console.WriteLine("Doing Work!");

            Console.WriteLine("Total Price is:" + totalPrice);
            Console.ReadKey();
        }
        //虚拟数据
        static IList<Order> GetOrderList()
        {
            IList<Order> orderList = new List<Order>();
            Order order1 = new Order();
            order1.ID = 1;
            orderList.Add(order1);
            ............
            return orderList;
        }
        //虚拟数据
        static IList<OrderItem> GetOrderItem()
        {
            IList<OrderItem> itemList = new List<OrderItem>();

            OrderItem orderItem1 = new OrderItem();
            orderItem1.ID = 1;
            orderItem1.Goods = "iPhone 4S";
            orderItem1.Price = 6700;
            orderItem1.Count = 2;
            orderItem1.OrderID = 1;
            itemList.Add(orderItem1);
            ...........
            return itemList;
        }

        //显示线程池现状
        static void ThreadPoolMessage(Order order)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("OrderID:{0}  OrderPrice:{1}\n" +
                  "  CurrentThreadId is {2}\n  WorkerThreads is:{3}" +
                  "  CompletionPortThreads is:{4}\n",
                  order.ID, order.Price,
                  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }
    }

运行结果

7.2.2 任务并行

在TPL当中还可以使用Parallel.Invoke方法触发多个异步任务,其中 actions 中可以包含多个方法或者委托，parallelOptions用于配置Parallel类的操作。
public static void Invoke(Action[] actions )
public static void Invoke(ParallelOptions parallelOptions, Action[] actions )
下面例子中利用了Parallet.Invoke并行查询多个Person，actions当中可以绑定方法、lambda表达式或者委托，注意绑定方法时必须是返回值为void的无参数方法。

class Program
    {
        static void Main(string[] args)
        {
            //设置最大线程数
            ThreadPool.SetMaxThreads(1000, 1000);
            
            //任务并行
            Parallel.Invoke(option,
                PersonMessage, 
                ()=>ThreadPoolMessage(GetPersonList()[1]),  
                delegate(){
                    ThreadPoolMessage(GetPersonList()[2]);
                });
            Console.ReadKey();
        }

        static void PersonMessage()
        {
            ThreadPoolMessage(GetPersonList()[0]);
        }

        //显示线程池现状
        static void ThreadPoolMessage(Person person)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
                  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
                  "  CompletionPortThreads is :{5}\n",
                  person.ID, person.Name, person.Age,
                  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

            Console.WriteLine(message);
        }

        //模拟源数据
        static IList<Person> GetPersonList()
        {
            var personList = new List<Person>();

            var person1 = new Person();
            person1.ID = 1;
            person1.Name = "Leslie";
            person1.Age = 30;
            personList.Add(person1);
            ..........
            return personList;
        }
    }

运行结果

7.3 Task简介

以Thread创建的线程被默认为前台线程，当然你可以把线程IsBackground属性设置为true，但TPL为此提供了一个更简单的类Task。
Task存在于System.Threading.Tasks命名空间当中，它可以作为异步委托的简单替代品。
通过Task的Factory属性将返回TaskFactory类，以TaskFactory.StartNew（Action）方法可以创建一个新线程，所创建的线程默认为后台线程。

class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(1000, 1000);
            Task.Factory.StartNew(() => ThreadPoolMessage());
            Console.ReadKey();
        }

        //显示线程池现状
        static void ThreadPoolMessage()
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("CurrentThreadId is:{0}\n" +
                "CurrentThread IsBackground:{1}\n" +
                "WorkerThreads is:{2}\nCompletionPortThreads is:{3}\n",
                 Thread.CurrentThread.ManagedThreadId,
                 Thread.CurrentThread.IsBackground.ToString(),
                 a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

运行结果

若要取消处理，可以利用CancellationTakenSource对象，在TaskFactory中包含有方法
public Task StartNew( Action action, CancellationToken cancellationToken )
在方法中加入CancellationTakenSource对象的CancellationToken属性，可以控制任务的运行，调用CancellationTakenSource.Cancel时任务就会自动停止。下面以图片下载为例子介绍一下TaskFactory的使用。

服务器端页面

<html xmlns="http://www.w3.org/1999/xhtml">
<head runat="server">
    <title></title>
    <script type="text/C#" runat="server">
        private static List<string> url=new List<string>();

        protected void Page_Load(object sender, EventArgs e)
        {
            if (!Page.IsPostBack)
            {
                url.Clear();
                Application["Url"] = null;
            }
        }

        protected void CheckBox_CheckedChanged(object sender, EventArgs e)
        {
            CheckBox checkBox = (CheckBox)sender;
            if (checkBox.Checked)
                url.Add(checkBox.Text);
            else
                url.Remove(checkBox.Text);
            Application["Url"]= url;
        }
</script>
</head>
<body>
    <form id="form1" runat="server" >
    <div align="left">
       <div align="center" style="float: left;">
         <asp:Image ID="Image1" runat="server" ImageUrl="~/Images/A.jpg" /><br />
         <asp:CheckBox ID="CheckBox1" runat="server" AutoPostBack="True" 
               oncheckedchanged="CheckBox_CheckedChanged" Text="A.jpg" />
       </div>
       <div align="center" style="float: left">
          <asp:Image ID="Image2" runat="server" ImageUrl="~/Images/B.jpg" /><br />
          <asp:CheckBox ID="CheckBox2" runat="server" AutoPostBack="True" 
               oncheckedchanged="CheckBox_CheckedChanged" Text="B.jpg" />
       </div>
       <div align="center" style="float: left">
          <asp:Image ID="Image3" runat="server" ImageUrl="~/Images/C.jpg" /><br />
          <asp:CheckBox ID="CheckBox3" runat="server" AutoPostBack="True" 
               oncheckedchanged="CheckBox_CheckedChanged" Text="C.jpg" />
       </div>
       <div align="center" style="float: left">
          <asp:Image ID="Image4" runat="server" ImageUrl="~/Images/D.jpg" /><br />
          <asp:CheckBox ID="CheckBox4" runat="server" AutoPostBack="True" 
               oncheckedchanged="CheckBox_CheckedChanged" Text="D.jpg" />
       </div>
       <div align="center" style="float: left">
          <asp:Image ID="Image5" runat="server" ImageUrl="~/Images/E.jpg" /><br />
          <asp:CheckBox ID="CheckBox5" runat="server" AutoPostBack="True" 
               oncheckedchanged="CheckBox_CheckedChanged" Text="E.jpg" />
       </div>        
    </div>
    </form>
</body>
</html>

首先在服务器页面中显示多个*.jpg图片，每个图片都有对应的CheckBox检测其选择情况。
所选择图片的路径会记录在Application["Url"]当中传递到Handler.ashx当中。

注意：Application是一个全局变量，此处只是为了显示Task的使用方式，在ASP.NET开发应该慎用Application。

Handler.ashx 处理图片的下载，它从 Application["Url"] 当中获取所选择图片的路径，并把图片转化成byte[]二进制数据。
再把图片的数量，每副图片的二进制数据的长度记录在OutputStream的头部。
最后把图片的二进制数据记入 OutputStream 一并输出。

public class Handler : IHttpHandler 
{
    public void ProcessRequest(HttpContext context)
    {
        //获取图片名，把图片数量写OutputStream
        List<String> urlList = (List<string>)context.Application["Url"];
        context.Response.OutputStream.Write(BitConverter.GetBytes(urlList.Count), 0, 4);
        
        //把图片转换成二进制数据
        List<string> imageList = GetImages(urlList);
        
        //把每副图片长度写入OutputStream
        foreach (string image in imageList)
        {
            byte[] imageByte=Convert.FromBase64String(image);
            context.Response.OutputStream.Write(BitConverter.GetBytes(imageByte.Length),0,4);
        }
        
        //把图片写入OutputStream
        foreach (string image in imageList)
        {
            byte[] imageByte = Convert.FromBase64String(image);
            context.Response.OutputStream.Write(imageByte,0,imageByte.Length);
        }
    }

    //获取多个图片的二进制数据
    private List<string> GetImages(List<string> urlList)
    {
        List<string> imageList = new List<string>();
        foreach (string url in urlList)
            imageList.Add(GetImage(url));
        return imageList;
    }
    
    //获取单副图片的二进制数据
    private string GetImage(string url)
    {
        string path = "E:/My Projects/Example/WebSite/Images/"+url;
        FileStream stream = new FileStream(path, FileMode.Open, FileAccess.Read);
        byte[] imgBytes = new byte[10240];
        int imgLength = stream.Read(imgBytes, 0, 10240);       
        return Convert.ToBase64String(imgBytes,0,imgLength);
    }

    public bool IsReusable
    {
        get{ return false;}
    }
}

客户端

建立一个WinForm窗口，里面加入一个WebBrowser连接到服务器端的Default.aspx页面。
当按下Download按键时，系统就会利用TaskFactory.StartNew的方法建立异步线程，使用WebRequest方法向Handler.ashx发送请求。
接收到回传流时，就会根据头文件的内容判断图片的数量与每副图片的长度，把二进制数据转化为*.jpg文件保存。

系统利用TaskFactory.StartNew(action,cancellationToken) 方式异步调用GetImages方法进行图片下载。
当用户按下Cancel按钮时，异步任务就会停止。值得注意的是，在图片下载时调用了CancellationToken.ThrowIfCancellationRequested方法，目的在检查并行任务的运行情况，在并行任务被停止时释放出OperationCanceledException异常，确保用户按下Cancel按钮时，停止所有并行任务。

public partial class Form1 : Form
    {
        private CancellationTokenSource tokenSource = new CancellationTokenSource();
        
        public Form1()
        {
            InitializeComponent();
            ThreadPool.SetMaxThreads(1000, 1000);
        }

        private void downloadToolStripMenuItem_Click(object sender, EventArgs e)
        {
             Task.Factory.StartNew(GetImages,tokenSource.Token);
        }

        private void cancelToolStripMenuItem_Click(object sender, EventArgs e)
        {
            tokenSource.Cancel();
        }

        private void GetImages()
        {
            //发送请求，获取输出流
            WebRequest webRequest = HttpWebRequest.Create("Http://localhost:5800/Handler.ashx"); 
            Stream responseStream=webRequest.GetResponse().GetResponseStream();

            byte[] responseByte = new byte[81960];
            IAsyncResult result=responseStream.BeginRead(responseByte,0,81960,null,null);
            int responseLength = responseStream.EndRead(result);

            //获取图片数量
            int imageCount = BitConverter.ToInt32(responseByte, 0);
            
            //获取每副图片的长度
            int[] lengths = new int[imageCount];
            for (int n = 0; n < imageCount; n++)
            {
                int length = BitConverter.ToInt32(responseByte, (n + 1) * 4);
                lengths[n] = length;
            }
            try
            {
                //保存图片
                for (int n = 0; n < imageCount; n++)
                {
                    string path = string.Format("E:/My Projects/Example/Test/Images/pic{0}.jpg", n);
                    FileStream file = new FileStream(path, FileMode.Create, FileAccess.ReadWrite);

                    //计算字节偏移量
                    int offset = (imageCount + 1) * 4;
                    for (int a = 0; a < n; a++)
                        offset += lengths[a];

                    file.Write(responseByte, offset, lengths[n]);
                    file.Flush();

                    //模拟操作
                    Thread.Sleep(1000);

                    //检测CancellationToken变化
                    tokenSource.Token.ThrowIfCancellationRequested();
                }
            }
            catch (OperationCanceledException ex)
            {
                MessageBox.Show("Download cancel!");
            }
        }
    }

7.4 并行查询（PLINQ）

并行 LINQ (PLINQ) 是 LINQ 模式的并行实现，主要区别在于 PLINQ 尝试充分利用系统中的所有处理器。 它利用所有处理器的方法，把数据源分成片段，然后在多个处理器上对单独工作线程上的每个片段并行执行查询， 在许多情况下，并行执行意味着查询运行速度显著提高。但这并不说明所有PLINQ都会使用并行方式，当系统测试要并行查询会对系统性能造成损害时，那将自动化地使用同步执行。
在System.Linq.ParallelEnumerable类中，包含了并行查询的大部分方法。

7.4.1 AsParallel

通常想要实现并行查询，只需向数据源添加 AsParallel 查询操作即可。

class Program
    {
        static void Main(string[] args)
        {
            var personList=GetPersonList().AsParallel() 
                   .Where(x=>x.Age>30);
            Console.ReadKey();
        }

        //模拟源数据
        static IList<Person> GetPersonList()
        {
            var personList = new List<Person>();

            var person1 = new Person();
            person1.ID = 1;
            person1.Name = "Leslie";
            person1.Age = 30;
            personList.Add(person1);
            ...........
            return personList;
        }
    }

7.4.2 AsOrdered

若要使查询结果必须保留源序列排序方式，可以使用AsOrdered方法。
AsOrdered依然使用并行方式，只是在查询过程加入额外信息，在并行结束后把查询结果再次进行排列。

class Program
    {
        static void Main(string[] args)
        {
            var personList=GetPersonList().AsParallel().AsOrdered()
                .Where(x=>x.Age<30);
            Console.ReadKey();
        }

        static IList<Person> GetPersonList()
        {......}
    }

7.4.3 WithDegreeOfParallelism

默认情况下，PLINQ 使用主机上的所有处理器，这些处理器的数量最多可达 64 个。
通过使用 WithDegreeOfParallelism(Of TSource) 方法，可以指示 PLINQ 使用不多于指定数量的处理器。

class Program
    {
        static void Main(string[] args)
        {
            var personList=GetPersonList().AsParallel().WithDegreeOfParallelism(2)
                .Where(x=>x.Age<30);
            Console.ReadKey();
        }

        static IList<Person> GetPersonList()
        {.........}
    }

7.4.4 ForAll

如果要对并行查询结果进行操作，一般会在for或foreach中执行，执行枚举操作时会使用同步方式。
有见及此，PLINQ中包含了ForAll方法，它可以使用并行方式对数据集进行操作。

class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(1000, 1000);
            GetPersonList().AsParallel().ForAll(person =>{
                ThreadPoolMessage(person);
            });
            Console.ReadKey();
        }

        static IList<Person> GetPersonList()
        {.......}

         //显示线程池现状
        static void ThreadPoolMessage(Person person)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
                  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
                  "  CompletionPortThreads is :{5}\n",
                  person.ID, person.Name, person.Age,
                  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

运行结果

7.4.5 WithCancellation

如果需要停止查询，可以使用 WithCancellation(Of TSource) 运算符并提供 CancellationToken 实例作为参数。
与第三节Task的例子相似，如果标记上的 IsCancellationRequested 属性设置为 true，则 PLINQ 将会注意到它，并停止所有线程上的处理，然后引发 OperationCanceledException。这可以保证并行查询能够立即停止。

class Program
    {
        static CancellationTokenSource tokenSource = new CancellationTokenSource();

        static void Main(string[] args)
        {
            Task.Factory.StartNew(Cancel);
            try
            {
                GetPersonList().AsParallel().WithCancellation(tokenSource.Token)
                    .ForAll(person =>
                    {
                        ThreadPoolMessage(person);
                    });
            }
            catch (OperationCanceledException ex)
            { }
            Console.ReadKey();
        }

        //在10~50毫秒内发出停止信号
        static void Cancel()
        {
            Random random = new Random();
            Thread.Sleep(random.Next(10,50));
            tokenSource.Cancel();
        }

        static IList<Person> GetPersonList()
        {......}

        //显示线程池现状
        static void ThreadPoolMessage(Person person)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
                  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
                  "  CompletionPortThreads is :{5}\n",
                  person.ID, person.Name, person.Age,
                  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

八、定时器与锁

8.1定时器

若要长期定时进行一些工作，比如像邮箱更新，实时收听信息等等，可以利用定时器Timer进行操作。
在System.Threading命名空间中存在Timer类与对应的TimerCallback委托，它可以在后台线程中执行一些长期的定时操作，使主线程不受干扰。
Timer类中最常用的构造函数为 public Timer( timerCallback , object , int , int )
timerCallback委托可以绑定执行方法，执行方法必须返回void，它可以是无参数方法，也可以带一个object参数的方法。
第二个参数是为 timerCallback 委托输入的参数对象。
第三个参数是开始执行前等待的时间。
第四个参数是每次执行之间的等待时间。

开发实例

class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(1000, 1000);

            TimerCallback callback = new TimerCallback(ThreadPoolMessage);
            Timer t = new Timer(callback,"Hello Jack! ", 0, 1000);
            Console.ReadKey();
        }

        //显示线程池现状
        static void ThreadPoolMessage(object data)
        {
            int a, b;
            ThreadPool.GetAvailableThreads(out a, out b);
            string message = string.Format("{0}\n   CurrentThreadId is:{1}\n" +
                "   CurrentThread IsBackground:{2}\n" +
                "   WorkerThreads is:{3}\n   CompletionPortThreads is:{4}\n",
                 data + "Time now is " + DateTime.Now.ToLongTimeString(),
                 Thread.CurrentThread.ManagedThreadId,
                 Thread.CurrentThread.IsBackground.ToString(),
                 a.ToString(), b.ToString());
            Console.WriteLine(message);
        }
    }

注意观察运行结果，每次调用Timer绑定的方法时不一定是使用同一线程，但线程都会是来自工作者线程的后台线程。

8.2 锁

在使用多线程开发时，存在一定的共用数据，为了避免多线程同时操作同一数据，.NET提供了lock、Monitor、Interlocked等多个锁定数据的方式。

8.2.1 lock

lock的使用比较简单，如果需要锁定某个对象时，可以直接使用lock(this)的方式。

private void Method()
{
      lock(this)
      {
          //在此进行的操作能保证在同一时间内只有一个线程对此对象操作
      }
}

如果操作只锁定某段代码，可以事先建立一个object对象，并对此对象进行操作锁定，这也是.net提倡的锁定用法。

class Control
{
      private object obj=new object();
      
      public void Method()
      {
            lock(obj)
            {.......}
      }
}

8.2.2 Montior

Montior存在于System.Thread命名空间内，相比lock，Montior使用更灵活。
它存在 Enter, Exit 两个方法，它可以对对象进行锁定与解锁，比lock使用更灵活。

class Control
{
      private object obj=new object();
 
      public void Method()
      {
            Monitor.Enter(obj);
            try
            {......}
            catch(Excetion ex)
            {......}
            finally
            {
                Monitor.Exit(obj);
            }
      }
}

使用try的方式，能确保程序不会因死锁而释放出异常！
而且在finally中释放obj对象能够确保无论是否出现死锁状态，系统都会释放obj对象。
而且Monitor中还存在Wait方法可以让线程等待一段时间，然后在完成时使用Pulse、PulseAll等方法通知等待线程。

8.2.3 Interlocked

Interlocked存在于System.Thread命名空间内，它的操作比Monitor使用更简单。
它存在CompareExchange、Decrement、Exchange、Increment等常用方法让参数在安全的情况进行数据交换。

Increment、Decrement 可以使参数安全地加1或减1并返回递增后的新值。

class Example
{
      private int a=1;

      public void AddOne()
      {
             int newA=Interlocked.Increment(ref a);
      }
}

xchange可以安全地变量赋值。

 public void SetData()
 {
       Interlocked.Exchange(ref a,100);
 }

CompareExchange使用特别方便，它相当于if的用法，当a等于1时，则把100赋值给a。

 public void CompareAndExchange() {
     Interlocked.CompareExchange(ref a,100,1);
 }

结束语

熟悉掌握多线程开发，对提高系统工作效率非常有帮助，尤其是回调方法与最近火热的并行编程更应该引起各位的重视。
在下用了较长的时间总结所用过的多线程开发方式，希望本篇文章能对各位的学习研究有所帮助，当中有所错漏的地方敬请点评