首页 > 代码库 > 线程池 异步I/O线程 <第三篇>

线程池 异步I/O线程 <第三篇>

在学习异步之前先来说说异步的好处,例如对于不需要CPU参数的输入输出操作,可以将实际的处理步骤分为以下三步:

  1. 启动处理;
  2. 实际的处理,此时不需要CPU参数;
  3. 任务完成后的处理;

  以上步骤如果仅仅使用一个线程,当线程正在处理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);

  参数说明:

  1. path是文件的相对路径或绝对路径;
  2. mode确定如何打开或创建文件;
  3. access确认访问文件的方式;
  4. share确定文件如何进程共享;
  5. bufferSize是代表缓冲区大小,一般默认最小值为8,在启动异步读取或写入时,文件大小一般大于缓冲大小;
  6. 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线程 <第三篇>