使用AsyncEnumerator简化异步操作

上一次提到了如何跨线程访问GUI。而这个需求往往是异步操作导致的。今天我们就来看看Jeffrey Richter写的AsyncEnumerator如何帮助我们处理异步问题。

先来看看最简单的一段异步下载网页的代码：

    public class Program
    {
        private static WebRequest request;

        public static void Main(string[] args)
        {
            request = WebRequest.Create("http://www.thoughtworks.com.cn");
            request.BeginGetResponse(HandleResponse, null);
            Console.ReadLine();
        }

        private static void HandleResponse(IAsyncResult ar)
        {
            request.EndGetResponse(ar);
            Console.WriteLine("Get response from web");
        }
    }

很简单不是吗？如果我们下载之后还要异步存储到本地的磁盘，这个时候就不是那么容易了：

    public class Program
    {
        private static WebRequest webRequest;
        private static FileStream fileStream;
        private static Stream responseStream;
        private static byte[] buffer;
        private static int count;

        public static void Main(string[] args)
        {
            webRequest = WebRequest.Create("http://www.thoughtworks.com.cn");
            webRequest.BeginGetResponse(HandleGetResponse, null);
            Console.ReadLine();
        }

        private static void HandleGetResponse(IAsyncResult ar)
        {
            var response = webRequest.EndGetResponse(ar);
            Console.WriteLine("Get response from web");

            responseStream = response.GetResponseStream();
            buffer = new byte[4096];
            BeginRead();
            fileStream = new FileStream(@"c:\downloaded.html", FileMode.Create);
        }

        private static void BeginRead()
        {
            responseStream.BeginRead(buffer, 0, buffer.Length, HandleReadResponseStream, null);
        }

        private static void HandleReadResponseStream(IAsyncResult ar)
        {
            Console.WriteLine("Read a chunk");
            count = responseStream.EndRead(ar);
            if (count == 0)
            {
                Console.WriteLine("Finished downloading from response");
                responseStream.Dispose();
                fileStream.Dispose();
                return;
            }
            fileStream.BeginWrite(buffer, 0, count, HandleWriteFileStream, null);
        }

        private static void HandleWriteFileStream(IAsyncResult ar)
        {
            Console.WriteLine("Write a chunk");
            fileStream.EndWrite(ar);
            BeginRead();
        }
    }

代码太长了，以至于我不得不折叠起来。这段代码还是有问题的，因为它没有处理异常情况，中途出个错，文件就不会被关闭。

从逻辑上来说获取Response，读取Response流，写入本地文件流从执行顺序上来说是一个完成之后，然后过一会儿下个接上执行。理论上来讲它就是

HandleGetResponse(xxx);
while(NotFinished(xxx)) {
  HandleReadResponseStream(xxx);
  HandleWriteFileStream(xxx);
}
CleanUp();

但是我们不能这么写。因为在每个操作之间都是一个异步等待的过程。实际上，是因为异步操作把一个完成的流程打散到了多个回调函数中去完成。那么有什么办法可以让一个方法执行一段，然后等一会，再执行一段呢？有，这就是yield。yield代表我暂时放弃执行的权利，等IO完成之后你再来执行我，我接着干下面的操作。

        private static void Demo()
        {
            int i = 1;
            foreach (var fib in Fib())
            {
                Console.WriteLine(i + ": " + fib);
                if (i++ > 10)
                {
                    break;
                }
            }
        }

        private static IEnumerable<int> Fib()
        {
            int i = 1;
            int j = 1;
            yield return i;
            yield return j;
            while (true)
            {
                var k = i + j;
                yield return k;
                i = j;
                j = k;
            }
        }

这个例子中，Fib（斐波那契额数列）是一个死循环。如果它是一个普通的函数，你是不能执行它的，因为它永远不会放弃执行权，它会一只拽着CPU去算终极的fib。但是我们这个例子中的Fib不会。它在每次yield return的时候，都会跳出函数，返回到调用的地方。然后每次调用，都会从上次执行的地方继续下去，继续执行的时候所有的局部状态（局部变量的值）都保留着上次的值。在foreach的背后是这么一个过程：

            var enumerable = Fib();
            var enumerator = enumerable.GetEnumerator();
            enumerator.MoveNext(); //Fib被执行 return i;
            Console.WriteLine(enumerator.Current);
            enumerator.MoveNext(); //Fib被继续执行 return j;
            Console.WriteLine(enumerator.Current);
            enumerator.MoveNext(); //Fib被继续执行 return i+j;
            Console.WriteLine(enumerator.Current);

所以我们只要把上面的IO操作序列，稍微改写就可以让它们不四处散落了：

BeginGetResponse(xxx);
yield return 1;
EndGetReponse(xxx);
while(NotFinished(xxx)) {
  BeginReadResponseStream(xxx);
  yield return 1;
  EndGetResponseStream(xxx);
  BeginWriteFileStream(xxx);
  yield return 1;
  EndGetResponseStream(xxx);
}
CleanUp();

因为每次yield return都会放弃执行权，所以我们可以在这个函数外的某处等待BeginXXX操作的回调，等IO操作完成了再来继续执行这个函数。基于这个想法（最开始是这位仁兄想出来的http://msmvps.com/blogs/mihailik/archive/2005/12/26/79813.aspx），Jeffrey Richter写了Power Threading库（wintellect上的下载链接坏了，用这个http://www.wintellect.com/Downloads/PowerThreadingAttachments/Wintellect_Power_Threading_Library_(May_15,_2008).zip）。最后的代码是这个样子的：

        private static IEnumerator<int> Download(AsyncEnumerator ae)
        {
            var webRequest = WebRequest.Create("http://www.thoughtworks.com.cn");
            webRequest.BeginGetResponse(ae.End(), null);
            yield return 1;
            var response = webRequest.EndGetResponse(ae.DequeueAsyncResult());
            Console.WriteLine("Get response from web");
            var buffer = new byte[4096];
            var count = buffer.Length;
            using (var responseStream = response.GetResponseStream())
            {
                using (var fileStream = new FileStream(@"c:\downloaded.html", FileMode.Create))
                {
                    while (count > 0)
                    {
                        Console.WriteLine("Read a chunk");
                        responseStream.BeginRead(buffer, 0, buffer.Length, ae.End(), null);
                        yield return 1;
                        count = responseStream.EndRead(ae.DequeueAsyncResult());
                        Console.WriteLine("Write a chunk");
                        fileStream.BeginWrite(buffer, 0, count, ae.End(), null);
                        yield return 1;
                        fileStream.EndWrite(ae.DequeueAsyncResult());
                    }
                }
            }
            Console.WriteLine("Finished downloading from response");
        }

是不是很简单呢？不过还有一个问题，那就是yield return我明白，是为了暂时退出这个函数，等待异步操作完成之后继续执行。但是我不明白的是，为什么是yield return 1呢？

其实这个yield return 1是给另外一个高级功能使用的。它的意思是“等待1个异步操作结束，然后执行我这行之后的代码“。如果yield return 2，就是等待两个异步操作。所以你必须先begin两个异步操作，然后yield return 2去等待。AsyncEnumerator还有返回值等高级功能，并且AsycnEnumerator内部使用了上文提到的AsyncOperationManager，所以在你的代码中可以安全地操作GUI不用害怕跨线程的问题。

参考资料：

Asynchronous iterators：

http://msmvps.com/blogs/mihailik/archive/2005/12/26/79813.aspx

Simplified APM With The AsyncEnumerator：

http://msdn.microsoft.com/en-us/magazine/cc546608.aspx

Simplified APM with C#：

http://msdn.microsoft.com/en-us/magazine/cc163323.aspx

More AsyncEnumerator Features：

http://msdn.microsoft.com/en-us/magazine/cc721613.aspx

Using C# 2.0 iterators to simplify writing asynchronous code：

http://blogs.msdn.com/michen/archive/2006/03/30/using-c-2-0-iterators-to-simplify-writing-asynchronous-code.aspx

http://blogs.msdn.com/michen/archive/2006/04/01/using-c-2-0-iterators-to-simplify-writing-asynchronous-code-part-2.aspx

附记：

喂，博主啊？为什么不直接创建一个新线程，然后在那里用同步操作完成上述动作？这个问题在我这里等价为什么要使用.NET的APM（Asynchronous Programming Model，异步编程模型）。正确的答案，参见http://msdn.microsoft.com/en-us/magazine/cc301191.aspx，Jeffrey Richter肯定写过这个问题的答案。不那么正确的答案：

1、提供如何实现异步操作的灵活性，新线程只是很多实现中的一种

这样我们可以利用Windows的Overlapped I/O，而这个就是一个内核级别的回调，不牵涉线程的问题了。性能直追epoll。

2、提供了何时使用新线程的灵活性，在一开始创建一个新线程然后把所有代码放到那里同步执行只是其中一种。

一个很流行的idea，叫SEDA（Staged Event Driven Architecture），究其核心就是把长操作分解成为异步的短操作，然后用不同大小的Thread Pool来回调不同类型的异步操作，通过调优达到线程在stage之间的最佳配比。这样避免了一有请求就起新线程的开销，线程多了系统就响应不过来了。又避免了单线程异步回调的低资源利用率，特别是CPU已经多核了的情况下。利用APM和AsyncEnumerator，再加上自己实现的ThreadPool，做一个.NET版本的SEDA架构也是可能的。