首页 > 代码库 > 线程池 异步I/O线程 <第三篇>
线程池 异步I/O线程 <第三篇>
在学习异步之前先来说说异步的好处,例如对于不需要CPU参数的输入输出操作,可以将实际的处理步骤分为以下三步:
- 启动处理;
- 实际的处理,此时不需要CPU参数;
- 任务完成后的处理;
以上步骤如果仅仅使用一个线程,当线程正在处理UI操作时就会出现“卡”的现象。
如果使用异步的处理方式,则这三步处理过程涉及到两个线程,主线程中启动第一步;第一步启动后,主线程结束(如果不结束,只会让该线程处于无作为的等待状态);第二步不需要CPU参与;第二步完成之后,在第二个线程上启动第三步;完成之后第二个线程结束。这样的处理过程中没有一个线程需要处于等待状态,使得运行的线程得到充分利用。
一、CLR线程池的I/O线程
上一篇学习的都是CLR线程池的辅助线程,这次要学习的是CLR线程池的I/O线程。
I/O线程是.NET专为访问外部资源所设置的一种线程,因为访问外部资源常常要受到外界因素的影响,为了防止让主线程受影响而长期处于阻塞状态,.NET为多个I/O操作都建立起了异步方法。例如:FileStream、TCP/IP、WebRequest、WebService等等,而且每个异步方法的使用方式都非常类似,
都是以Beginxxx开始,以Endxxx结束(APM)。对于APM来说,必须使用Endxxx结束异步,否则可能会造成资源泄露。Beginxxx实际是将线程排入线程池。
另外还有一种基于事件的异步编程模式(EPM),支持基于事件的异步模式的类将有一个或多个后缀为Async的方法,同时还会有一个相应名为Completed后缀的事件,Async方法用于启动异步处理,而Completed事件将在异步处理完成之后通过事件来宣告异步处理的完成。注意,在使用EPM模式的时候,不管是完成了异步请求,还是在处理中出现异常,或者是终止异步处理,都必须要调用Compeleted处理程序。如:
OpenReadAsync OpenReadCompleted
二、异步读写FileStream
需要在FileStream中异步调用I/O线程,必须使用以下构造函数建立FileStream对象,并把useAsync设置为true。
FileStream stream = new FileStream(string path,FileMode mode,FileAccess access,FileShare share,int bufferSize,bool useAsync);
参数说明:
- path是文件的相对路径或绝对路径;
- mode确定如何打开或创建文件;
- access确认访问文件的方式;
- share确定文件如何进程共享;
- bufferSize是代表缓冲区大小,一般默认最小值为8,在启动异步读取或写入时,文件大小一般大于缓冲大小;
- userAsync代表是否启动异步I/O线程。
注意:当使用BeginRead和BeginWrite方法在执行大量读或写时效果更好,但对于少量读/写,这些方法速度可能比同步还要慢,因为进行线程间的切换需要大量时间。
1、异步写入
FileStream中包含BeginWrite、EndWrite方法可以启动I/O线程进行异步写入。
public override IAsyncResult BeginWrite(byte[] array,int offset,int numBytes,AsyncCallback,Object stateObject) public override void EndWrite(IAsyncResult asyncResult)
BeginWrite返回值为IAsyncResult,使用方式与委托的BeginInvoke方法相似,最好就是使用回调函数,避免线程阻塞。
最后两个参数还是同样的套路:
- AsyncCallback用于绑定回调函数;
- Object用于传递外部数据。
要注意一点:AsyncCallback所绑定的回调函数必须是带单个IAsyncResult参数的无返回值方法。
在例子中,把FileStream作为外部数据传递到回调函数当中,然后再回调函数中利用IAsyncResult.AsyncState获取FileStream对象,最后通过FileStream.EndWrite(IAsyncResult)结束写入。
下面是一个异步写入的例子:
class Program { static void Main(string[] args) { int a, b; ThreadPool.GetMaxThreads(out a, out b); Console.WriteLine("原有辅助线程数" + a + " " + "原有I/O线程数" + b); //文件名 文件创建方式 文件权限 文件进程共享 缓冲区大小为1024 是否启动异步I/O线程为true FileStream stream = new FileStream(@"D:\123.txt", FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.ReadWrite, 1024, true); //这里要注意,如果写入的字符串很小,则.Net会使用辅助线程写,因为这样比较快 byte[] bytes = Encoding.UTF8.GetBytes("你在他乡还好吗?"); //异步写入开始,倒数第二个参数指定回调函数,最后一个参数将自身传到回调函数里,用于结束异步线程 stream.BeginWrite(bytes, 0, (int)bytes.Length, new AsyncCallback(Callback), stream); ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("现有辅助线程数" + a + " " + "现有有I/O线程数" + b); Console.WriteLine("主线程继续干其他活!"); Console.ReadKey(); } static void Callback(IAsyncResult result) { //显示线程池现状 Thread.Sleep(2000); //通过result.AsyncState再强制转换为FileStream就能够获取FileStream对象,用于结束异步写入 FileStream stream = (FileStream)result.AsyncState; stream.EndWrite(result); stream.Flush(); stream.Close(); } }
输出结果如下:
对于结束异步线程的方法,还是玩IAsyncResult的这一套,在启动异步写时将自身对象传递到回调函数中,在回调函数中获得自身去结束异步线程。
这就是C#中的异步操作,从剩余线程数我们看到,异步实际上是调用线程池的线程来实现异步的。
2、异步读取
FileStream中可以通过使用BeginRead和EndRead调用异步I/O线程读取:
public override IAsyncResult BeginRead(byte[] array,int offset,int numBytes,AsyncCallback userCallback,Object stateObject) public override int EndRead(IAsyncResult asyncResult)
BeginRead与EndRead方法与写相似,AsyncCallback用于绑定回调函数;Object用于传递外部数据。在回调函数只需要使用IAsyncResult.AsyncState就可以获取外部数据。EndRead方法会返回从流中读取到的字节数量。
首先定义FileData类,里面包含FileStream对象,byte[]数组和长度。然后把FileData对象作为外部数据传到回调函数,在回调函数中,把IAsyncResult.AsyncState强制转换为FileData。然后通过FileStream.EndRead(IAsyncResult)结束读取。
最后比较一下长度,如果读取到的长度与输入的数据长度不一致,则抛出异常。
class Program { static void Main(string[] args) { int a, b; ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("原有辅助线程:" + a + "原有I/O线程:" + b); byte[] byteData = http://www.mamicode.com/new byte[1024];"D:\123.txt", FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.ReadWrite, 1024, true); //把FileStream对象,byte[]对象,长度等有关数据绑定到FileDate对象中,以附带属性方式送到回调函数 Hashtable ht = new Hashtable(); ht.Add("Length", (int)stream.Length); ht.Add("Stream", stream); ht.Add("ByteData", byteData); //启动异步读取,倒数第二个参数是指定回调函数,倒数第一个参数是传入回调函数中的参数 stream.BeginRead(byteData, 0, (int)ht["Length"], new AsyncCallback(Completed), ht); ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("现有辅助线程:" + a + "现有I/O线程:" + b); Console.ReadKey(); } //实际参数就是回调函数 static void Completed(IAsyncResult result) { Thread.Sleep(2000); //参数result实际上就是Hashtable对象,以FileStream.EndRead完成异步读取 Hashtable ht = (Hashtable)result.AsyncState; FileStream stream = (FileStream)ht["Stream"]; int length = stream.EndRead(result); stream.Close(); string str = Encoding.UTF8.GetString(ht["ByteData"] as byte[]); Console.WriteLine(str); stream.Close(); } }
输出如下:
注意,如果文件过小,小于缓冲区1024,那么可能会调用工作者线程而非I/O线程操作。但是根据我的观察,只是读取文件时文件过小可能会调用辅助线程操作,但是写入时不会。
像上面就是直接用辅助线程处理了。
IAsyncResult的作用主要有两点:
- AsyncState属性,用来传递参数到回调函数;
- Endxxx方法,结束异步操作方法需要此对象作为参数;
三、异步WebRequest
System.Net.WebRequest是.NET为实现Internet的"请求/响应模型"而开发的一个abstract基类。它主要有三个子类:
- FtpWebRequest,FileWebRequest使用"file://路径"的URI方式实现对本地资源和内部文件的请求/响应;
- HttpWebRequest,FtpWebRequest使用FTP文件传输协议实现文件请求/响应;
- FileWebRequest,HttpWebRequest用于处理HTTP的页面请求/响应;
当使用WebRequest.Create(string uri)创建对象时,应用程序就可以根据请求协议判断实现类来进行操作FileWebRequest、FtpWebRequest、HttpWebRequest各有其作用。由于使用方法类似,下面就用常用的HttpWebRequest为例子介绍一下异步WebRequest的使用方法。
HttpWebRequest包含由一下几个常用方法处理请求/响应:
public override Stream GetRequest() 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方法在于同一线程内。
class Program { static void Main(string[] args) { int a, b; ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("原有辅助线程:" + a + "原有I/O线程:" + b); //使用WebRequest.Create方法建立HttpWebRequest对象 HttpWebRequest webRequest = (HttpWebRequest)WebRequest.Create("http://www.baidu.com"); webRequest.Method = "post"; //对写入数据的RequestStream对象进行异步请求 IAsyncResult result = webRequest.BeginGetResponse(new AsyncCallback(EndGetResponse), webRequest); Thread.Sleep(1000); ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("现有辅助线程:" + a + "现有I/O线程:" + b); Console.WriteLine("主线程继续干其他事!"); Console.ReadKey(); } static void EndGetResponse(IAsyncResult result) { Thread.Sleep(2000); //结束异步请求,获取结果 HttpWebRequest webRequest = (HttpWebRequest)result.AsyncState; WebResponse webResponse = webRequest.EndGetResponse(result); Stream stream = webResponse.GetResponseStream(); StreamReader sr = new StreamReader(stream); string html = sr.ReadToEnd(); Console.WriteLine(html.Substring(0,50)); } }
显示结果如下:
四、异步SqlCommand
使用异步SqlCommand的时候,请注意把ConnectionString 的 Asynchronous Processing 设置为 true 。
class Program { static void Main(string[] args) { int a, b; ThreadPool.GetMaxThreads(out a, out b); Console.WriteLine("原有辅助线程数" + a + " " + "原有I/O线程数" + b); string str = "server=.;database=Test;uid=sa;pwd=123;Asynchronous Processing=true"; SqlConnection conn = new SqlConnection(str); SqlCommand cmd = conn.CreateCommand(); cmd.CommandText = "INSERT INTO Person VALUES(15,‘郭嘉‘,22)"; conn.Open(); cmd.BeginExecuteNonQuery(new AsyncCallback(EndCallback), cmd); Thread.Sleep(1000); ThreadPool.GetAvailableThreads(out a, out b); Console.WriteLine("现有辅助线程数" + a + " " + "现有I/O线程数" + b); Console.WriteLine("主线程继续执行!"); Console.ReadKey(); } public static void EndCallback(IAsyncResult result) { Thread.Sleep(2000); SqlCommand cmd = result.AsyncState as SqlCommand; //获得异步传入的参数 Console.WriteLine("成功执行命令:" + cmd.CommandText); Console.WriteLine("本次执行影响行数为:" + cmd.EndExecuteNonQuery(result)); cmd.Connection.Close(); } }
输出如下:
线程池 异步I/O线程 <第三篇>