F#中的异步及并行模式：反馈进度的事件

这里我们会使用这个设计模式开发一个示例，分析F#函数式编程语言中的反馈进度事件，其中部分示例代码来自于F# JAOO Tutorial。

创新互联公司-专业网站定制、快速模板网站建设、高性价比枣庄网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式枣庄网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖枣庄地区。费用合理售后完善，十年实体公司更值得信赖。

我们先来看一下这个设计模式的一个基础示例。在下面的代码中，我们会定义一个对象，以此来协调一组同时执行的异步任务。每个任务在结束之后会主动汇报它的结果，而不是等待统一的收集过程。

 
 
 
    
  
  
  
  1. type AsyncWorker<'T>(jobs: seq>) =  
    2. 
  
  
  
  2.    
    3. 
  
  
  
  3.     // This declares an F# event that we can raise  
    4. 
  
  
  
  4.     let jobCompleted  = new Event()  
    5. 
  
  
  
  5.    
    6. 
  
  
  
  6.     /// Start an instance of the work  
    7. 
  
  
  
  7.     member x.Start()    =  
    8. 
  
  
  
  8.         // Capture the synchronization context to allow us to raise events back on the GUI thread  
    9. 
  
  
  
  9.         let syncContext = SynchronizationContext.CaptureCurrent()  
    10. 
  
  
  
  10.    
    11. 
  
  
  
  11.         // Mark up the jobs with numbers  
    12. 
  
  
  
  12.         let jobsjobs = jobs |> Seq.mapi (fun i job -> (job,i+1))  
    13. 
  
  
  
  13.    
    14. 
  
  
  
  14.         let work =   
    15. 
  
  
  
  15.             Async.Parallel  
    16. 
  
  
  
  16.                [ for (job,jobNumber) in jobs -> 
    17. 
  
  
  
  17.                    async { let! result = job 
    18. 
  
  
  
  18.                            syncContext.RaiseEvent jobCompleted (jobNumber,result)  
    19. 
  
  
  
  19.                            return result } ]  
    20. 
  
  
  
  20.    
    21. 
  
  
  
  21.         Async.Start(work |> Async.Ignore)  
    22. 
  
  
  
  22.    
    23. 
  
  
  
  23.     /// Raised when a particular job completes  
    24. 
  
  
  
  24.     member x.JobCompleted  = jobCompleted.Publish 
    25.

设计模式的一些关键之处已经使用黄色进行高亮：在对象的Start方法中，我们在GUI线程中捕获了当前的“同步上下文”，这使得我们可以从GUI的上下文中运行代码或触发事件。我们还定义了一个私有的辅助函数来触发任意的F#事件，这虽不必须但可以使我们的代码变的更为整洁。定义了多个事件。这些事件作为属性发布，如果该对象还需要被其他.NET语言使用，则为它标记一个[]属性。

我们这里通过指定一个定义了任务内容的异步工作流来启动后台任务。Async.Start可以用来启动这个工作流（虽然Async.StartWithContinuations更为常用，例如在后面的示例中）。在后台任务产生进度之后，便会在合适的时候触发这些事件。这段代码使用了两个基于System.Threading.SynchronizationContext的辅助方法，它们会在这个系列的文章中多次出现。如下：

 
 
 
    
  
  
  
  1. type SynchronizationContext with  
    2. 
  
  
  
  2.     /// A standard helper extension method to raise an event on the GUI thread  
    3. 
  
  
  
  3.     member syncContext.RaiseEvent (event: Event<_>) args =  
    4. 
  
  
  
  4.         syncContext.Post((fun _ -> event.Trigger args),state=null)  
    5. 
  
  
  
  5.    
    6. 
  
  
  
  6.     /// A standard helper extension method to capture the current synchronization context.  
    7. 
  
  
  
  7.     /// If none is present, use a context that executes work in the thread pool.  
    8. 
  
  
  
  8.     static member CaptureCurrent () =  
    9. 
  
  
  
  9.         match SynchronizationContext.Current with  
    10. 
  
  
  
  10.         | null -> new SynchronizationContext()  
    11. 
  
  
  
  11.         | ctxt -> ctxt  
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13. 您现在便可以使用这个组件来管理一系列CPU密集型异步任务：   
    14. 
  
  
  
  14.  
    15. 
  
  
  
  15. let rec fib i = if i < 2 then 1 else fib (i-1) + fib (i-2)  
    16. 
  
  
  
  16.      
    17. 
  
  
  
  17.     let worker =  
    18. 
  
  
  
  18.         new AsyncWorker<_>( [ for i in 1 .. 100 -> async { return fib (i % 40) } ] )  
    19. 
  
  
  
  19.    
    20. 
  
  
  
  20.     worker.JobCompleted.Add(fun (jobNumber, result) -> 
    21. 
  
  
  
  21.         printfn "job %d completed with result %A" jobNumber result)  
    22. 
  
  
  
  22.    
    23. 
  
  
  
  23.     worker.Start()  
    24.

在执行时，每个任务结束之后便会汇报结果：

 
 
 
    
  
  
  
  1. job 1 completed with result 1  
    2. 
  
  
  
  2. job 2 completed with result 2  
    3. 
  
  
  
  3. ...  
    4. 
  
  
  
  4. job 39 completed with result 102334155  
    5. 
  
  
  
  5. job 77 completed with result 39088169  
    6. 
  
  
  
  6. job 79 completed with result 102334155 
    7.

我们可以使用多种方式让后台运行的任务汇报结果。在90%的情况下最简单的便是上面的方法：在GUI（或ASP.NET的Page_Load）线程中触发.NET事件。这个技巧隐藏了后台线程的使用细节，并利用了所有.NET程序员都非常熟悉的标准.NET惯例，以此保证用于实现并行编程的技术都得到了有效的封装。

汇报异步I/O的进度

反馈进度的事件模式也可以用在异步I/O操作上。例如这里有一系列I/O任务：

 
 
 
    
  
  
  
  1. open System.IO  
    2. 
  
  
  
  2. open System.Net  
    3. 
  
  
  
  3. open Microsoft.FSharp.Control.WebExtensions  
    4. 
  
  
  
  4.  
    5. 
  
  
  
  5. /// Fetch the contents of a web page, asynchronously.  
    6. 
  
  
  
  6. let httpAsync(url:string) =  
    7. 
  
  
  
  7.     async { let req = WebRequest.Create(url)  
    8. 
  
  
  
  8.             use! resp = req.AsyncGetResponse()  
    9. 
  
  
  
  9.             use stream = resp.GetResponseStream()  
    10. 
  
  
  
  10.             use reader = new StreamReader(stream)  
    11. 
  
  
  
  11.             let text = reader.ReadToEnd()  
    12. 
  
  
  
  12.             return text }  
    13. 
  
  
  
  13.  
    14. 
  
  
  
  14. let urls =  
    15. 
  
  
  
  15.     [ "http://www.live.com";  
    16. 
  
  
  
  16.       "http://news.live.com";  
    17. 
  
  
  
  17.       "http://www.yahoo.com";  
    18. 
  
  
  
  18.       "http://news.yahoo.com";  
    19. 
  
  
  
  19.       "http://www.google.com";  
    20. 
  
  
  
  20.       "http://news.google.com"; ]  
    21. 
  
  
  
  21.  
    22. 
  
  
  
  22. let jobs =  [ for url in urls -> httpAsync url ]  
    23. 
  
  
  
  23.  
    24. 
  
  
  
  24. let worker = new AsyncWorker<_>(jobs)  
    25. 
  
  
  
  25. worker.JobCompleted.Add(fun (jobNumber, result) -> 
    26. 
  
  
  
  26.     printfn "job %d completed with result %A" jobNumber result.Length)  
    27. 
  
  
  
  27.  
    28. 
  
  
  
  28. worker.Start()  
    29.

在执行过程中便会反馈进度结果，表现为每个Web页面的长度：

 
 
 
    
  
  
  
  1. job 5 completed with result 8521  
    2. 
  
  
  
  2. job 6 completed with result 155767  
    3. 
  
  
  
  3. job 3 completed with result 117778  
    4. 
  
  
  
  4. job 1 completed with result 16490  
    5. 
  
  
  
  5. job 4 completed with result 175186  
    6. 
  
  
  
  6. job 2 completed with result 70362 
    7.

#p#
反馈多种不同事件的任务

在这个设计模式中，我们使用了一个对象来封装和监督异步组合任务的执行过程，即使我们需要丰富API，也可以轻松地添加多个事件。例如，以下的代码添加了额外的事件来表示所有的任务已经完成了，或是其中某个任务出现了错误，还有便是整个组合完成之前便成功地取消了任务。以下高亮的代码便展示了事件的声明，触发及发布：

 
 
 
    
  
  
  
  1. open System  
    2. 
  
  
  
  2. open System.Threading  
    3. 
  
  
  
  3. open System.IO  
    4. 
  
  
  
  4. open Microsoft.FSharp.Control.WebExtensions  
    5. 
  
  
  
  5.    
    6. 
  
  
  
  6. type AsyncWorker<'T>(jobs: seq>) =  
    7. 
  
  
  
  7.    
    8. 
  
  
  
  8.            
    9. 
  
  
  
  9.     // Each of these lines declares an F# event that we can raise  
    10. 
  
  
  
  10.     let allCompleted  = new Event<'T[]>()  
    11. 
  
  
  
  11.     let error         = new Event()  
    12. 
  
  
  
  12.     let canceled      = new Event()  
    13. 
  
  
  
  13.     let jobCompleted  = new Event()  
    14. 
  
  
  
  14.    
    15. 
  
  
  
  15.     let cancellationCapability = new CancellationTokenSource()  
    16. 
  
  
  
  16.    
    17. 
  
  
  
  17.     /// Start an instance of the work  
    18. 
  
  
  
  18.     member x.Start()    =                                                         
    19. 
  
  
  
  19.                                                         
    20. 
  
  
  
  20.         // Capture the synchronization context to allow us to raise events back on the GUI thread  
    21. 
  
  
  
  21.         let syncContext = SynchronizationContext.CaptureCurrent()  
    22. 
  
  
  
  22.    
    23. 
  
  
  
  23.         // Mark up the jobs with numbers  
    24. 
  
  
  
  24.         let jobsjobs = jobs |> Seq.mapi (fun i job -> (job,i+1))  
    25. 
  
  
  
  25.    
    26. 
  
  
  
  26.         let work =   
    27. 
  
  
  
  27.             Async.Parallel  
    28. 
  
  
  
  28.                [ for (job,jobNumber) in jobs -> 
    29. 
  
  
  
  29.                    async { let! result = job 
    30. 
  
  
  
  30.                            syncContext.RaiseEvent jobCompleted (jobNumber,result)  
    31. 
  
  
  
  31.                            return result } ]  
    32. 
  
  
  
  32.    
    33. 
  
  
  
  33.         Async.StartWithContinuations  
    34. 
  
  
  
  34.             ( work,  
    35. 
  
  
  
  35.               (fun res -> raiseEventOnGuiThread allCompleted res),  
    36. 
  
  
  
  36.               (fun exn -> raiseEventOnGuiThread error exn),  
    37. 
  
  
  
  37.               (fun exn -> raiseEventOnGuiThread canceled exn ),  
    38. 
  
  
  
  38.              cancellationCapability.Token)  
    39. 
  
  
  
  39.    
    40. 
  
  
  
  40.     member x.CancelAsync() =  
    41. 
  
  
  
  41.        cancellationCapability.Cancel()  
    42. 
  
  
  
  42.          
    43. 
  
  
  
  43.     /// Raised when a particular job completes  
    44. 
  
  
  
  44.     member x.JobCompleted  = jobCompleted.Publish  
    45. 
  
  
  
  45.     /// Raised when all jobs complete  
    46. 
  
  
  
  46.     member x.AllCompleted  = allCompleted.Publish  
    47. 
  
  
  
  47.     /// Raised when the composition is cancelled successfully  
    48. 
  
  
  
  48.     member x.Canceled   = canceled.Publish  
    49. 
  
  
  
  49.     /// Raised when the composition exhibits an error  
    50. 
  
  
  
  50.     member x.Error      = error.Publish我们可以使用最普通的做法来响应这些额外的事件，例如：   
    51. 
  
  
  
  51.  
    52. 
  
  
  
  52. let worker = new AsyncWorker<_>(jobs)  
    53. 
  
  
  
  53.  
    54. 
  
  
  
  54. worker.JobCompleted.Add(fun (jobNumber, result) -> 
    55. 
  
  
  
  55.     printfn "job %d completed with result %A" jobNumber result.Length)  
    56. 
  
  
  
  56.  
    57. 
  
  
  
  57. worker.AllCompleted.Add(fun results -> 
    58. 
  
  
  
  58.     printfn "all done, results = %A" results )  
    59. 
  
  
  
  59.  
    60. 
  
  
  
  60. worker.Start()  
    61.

如上，这个监视中异步工作流可以支持任务的取消操作。反馈进度的事件模式可用于相当部分需要全程汇报进度的场景。在下一个示例中，我们使用这个模式来封装后台对于一系列Twitter采样消息的读取操作。运行这个示例需要一个Twitter帐号和密码。在这里只会发起一个事件，如果需要的话您也可以在某些情况下发起更多事件。F# JAOO Tutorial中也包含了这个示例。

 
 
 
    
  
  
  
  1. // F# Twitter Feed Sample using F# Async Programming and Event processing  
    2. 
  
  
  
  2. //  
    3. 
  
  
  
  3.    
    4. 
  
  
  
  4. #r "System.Web.dll"  
    5. 
  
  
  
  5. #r "System.Windows.Forms.dll"  
    6. 
  
  
  
  6. #r "System.Xml.dll"  
    7. 
  
  
  
  7.    
    8. 
  
  
  
  8. open System  
    9. 
  
  
  
  9. open System.Globalization  
    10. 
  
  
  
  10. open System.IO  
    11. 
  
  
  
  11. open System.Net  
    12. 
  
  
  
  12. open System.Web  
    13. 
  
  
  
  13. open System.Threading  
    14. 
  
  
  
  14. open Microsoft.FSharp.Control.WebExtensions  
    15. 
  
  
  
  15.    
    16. 
  
  
  
  16. /// A component which listens to tweets in the background and raises an  
    17. 
  
  
  
  17. /// event each time a tweet is observed  
    18. 
  
  
  
  18. type TwitterStreamSample(userName:string, password:string) =  
    19. 
  
  
  
  19.    
    20. 
  
  
  
  20.     let tweetEvent = new Event<_>()  
    21. 
  
  
  
  21.     let streamSampleUrl = "http://stream.twitter.com/1/statuses/sample.xml?delimited=length" 
    22. 
  
  
  
  22.    
    23. 
  
  
  
  23.     /// The cancellation condition  
    24. 
  
  
  
  24.     let mutable group = new CancellationTokenSource()  
    25. 
  
  
  
  25.    
    26. 
  
  
  
  26.     /// Start listening to a stream of tweets  
    27. 
  
  
  
  27.     member this.StartListening() =  
    28. 
  
  
  
  28.                                                          
    29. 
  
  
  
  29.         // Capture the synchronization context to allow us to raise events back on the GUI thread  
    30. 
  
  
  
  30.           
    31. 
  
  
  
  31.         // Capture the synchronization context to allow us to raise events back on the GUI thread  
    32. 
  
  
  
  32.         let syncContext = SynchronizationContext.CaptureCurrent()  
    33. 
  
  
  
  33.    
    34. 
  
  
  
  34.         /// The background process  
    35. 
  
  
  
  35.         let listener (syncContext: SynchronizationContext) =  
    36. 
  
  
  
  36.             async { let credentials = NetworkCredential(userName, password)  
    37. 
  
  
  
  37.                     let req = WebRequest.Create(streamSampleUrl, Credentials=credentials)  
    38. 
  
  
  
  38.                     use! resp = req.AsyncGetResponse()  
    39. 
  
  
  
  39.                     use stream = resp.GetResponseStream()  
    40. 
  
  
  
  40.                     use reader = new StreamReader(stream)  
    41. 
  
  
  
  41.                     let atEnd = reader.EndOfStream  
    42. 
  
  
  
  42.                     let rec loop() =  
    43. 
  
  
  
  43.                         async {  
    44. 
  
  
  
  44.                             let atEnd = reader.EndOfStream  
    45. 
  
  
  
  45.                             if not atEnd then  
    46. 
  
  
  
  46.                                 let sizeLine = reader.ReadLine()  
    47. 
  
  
  
  47.                                 let size = int sizeLine  
    48. 
  
  
  
  48.                                 let buffer = Array.zeroCreate size  
    49. 
  
  
  
  49.                                 let _numRead = reader.ReadBlock(buffer,0,size)   
    50. 
  
  
  
  50.                                 let text = new System.String(buffer)  
    51. 
  
  
  
  51.                                 syncContext.RaiseEvent tweetEvent text  
    52. 
  
  
  
  52.                                 return! loop()  
    53. 
  
  
  
  53.                         }  
    54. 
  
  
  
  54.                     return! loop() }  
    55. 
  
  
  
  55.    
    56. 
  
  
  
  56.         Async.Start(listener, group.Token)  
    57. 
  
  
  
  57.    
    58. 
  
  
  
  58.     /// Stop listening to a stream of tweets  
    59. 
  
  
  
  59.     member this.StopListening() =  
    60. 
  
  
  
  60.         group.Cancel();  
    61. 
  
  
  
  61.         group <- new CancellationTokenSource()  
    62. 
  
  
  
  62.    
    63. 
  
  
  
  63.     /// Raised when the XML for a tweet arrives  
    64. 
  
  
  
  64.     member this.NewTweet = tweetEvent.Publish在Twitter的标准采样消息流中每出现一条消息便会触发一个事件，并同时提供消息的内容。我们可以这样监听事件流：   
    65. 
  
  
  
  65.  
    66. 
  
  
  
  66. let userName = "..." // set Twitter user name here  
    67. 
  
  
  
  67. let password = "..." // set Twitter user name here  
    68. 
  
  
  
  68.    
    69. 
  
  
  
  69. let twitterStream = new TwitterStreamSample(userName, password)  
    70. 
  
  
  
  70.    
    71. 
  
  
  
  71. twitterStream.NewTweet  
    72. 
  
  
  
  72.    |> Event.add (fun s -> printfn "%A" s)  
    73. 
  
  
  
  73.    
    74. 
  
  
  
  74. twitterStream.StartListening()  
    75. 
  
  
  
  75. twitterStream.StopListening()  
    76.

#p#
程序运行后便会不断打印出每条消息的XML数据。您可以从Twitter API页面中来了解采样消息流的使用方式。如果您想同时解析这些消息，以下便是这一工作的示例代码。不过，也请关注Twitter API页面中的指导准则。例如，如果需要构建一个高可靠性的系统，您最好在处理前进行保存，或是使用消息队列。

 
 
 
    
  
  
  
  1. #r "System.Xml.dll"  
    2. 
  
  
  
  2. #r "System.Xml.Linq.dll"  
    3. 
  
  
  
  3. open System.Xml  
    4. 
  
  
  
  4. open System.Xml.Linq  
    5. 
  
  
  
  5.    
    6. 
  
  
  
  6. let xn (s:string) = XName.op_Implicit s  
    7. 
  
  
  
  7.    
    8. 
  
  
  
  8. /// The results of the parsed tweet  
    9. 
  
  
  
  9. type UserStatus =  
    10. 
  
  
  
  10.     { UserName : string  
    11. 
  
  
  
  11.       ProfileImage : string  
    12. 
  
  
  
  12.       Status : string  
    13. 
  
  
  
  13.       StatusDate : DateTime }  
    14. 
  
  
  
  14.    
    15. 
  
  
  
  15. /// Attempt to parse a tweet  
    16. 
  
  
  
  16. let parseTweet (xml: string) =  
    17. 
  
  
  
  17.    
    18. 
  
  
  
  18.     let document = XDocument.Parse xml  
    19. 
  
  
  
  19.      
    20. 
  
  
  
  20.     let node = document.Root  
    21. 
  
  
  
  21.     if node.Element(xn "user") <> null then  
    22. 
  
  
  
  22.         Some { UserName     = node.Element(xn "user").Element(xn "screen_name").Value;  
    23. 
  
  
  
  23.                ProfileImage = node.Element(xn "user").Element(xn "profile_image_url").Value;  
    24. 
  
  
  
  24.                Status       = node.Element(xn "text").Value       |> HttpUtility.HtmlDecode;  
    25. 
  
  
  
  25.                StatusDate   = node.Element(xn "created_at").Value |> (fun msg -> 
    26. 
  
  
  
  26.                                    DateTime.ParseExact(msg, "ddd MMM dd HH:mm:ss +0000 yyyy",  
    27. 
  
  
  
  27.                                                        CultureInfo.CurrentCulture)); }  
    28. 
  
  
  
  28.     else  
    29. 
  
  
  
  29.  
    30. 
  
  
  
  30. None基于事件流还可以使用组合式的编程：   
    31. 
  
  
  
  31.  
    32. 
  
  
  
  32. twitterStream.NewTweet  
    33. 
  
  
  
  33.    |> Event.choose parseTweet  
    34. 
  
  
  
  34.    |> Event.add (fun s -> printfn "%A" s)  
    35. 
  
  
  
  35.    
    36. 
  
  
  
  36. twitterStream.StartListening()或是收集统计数据：   
    37. 
  
  
  
  37.  
    38. 
  
  
  
  38. let addToMultiMap key x multiMap =  
    39. 
  
  
  
  39.    let prev = match Map.tryFind key multiMap with None -> [] | Some v -> v  
    40. 
  
  
  
  40.    Map.add x.UserName (x::prev) multiMap  
    41. 
  
  
  
  41.    
    42. 
  
  
  
  42. /// An event which triggers on every 'n' triggers of the input event  
    43. 
  
  
  
  43. let every n (ev:IEvent<_>) =  
    44. 
  
  
  
  44.    let out = new Event<_>()  
    45. 
  
  
  
  45.    let count = ref 0  
    46. 
  
  
  
  46.    ev.Add (fun arg -> incr count; if !count % n = 0 then out.Trigger arg)  
    47. 
  
  
  
  47.    out.Publish  
    48. 
  
  
  
  48.    
    49. 
  
  
  
  49. twitterStream.NewTweet  
    50. 
  
  
  
  50.    |> Event.choose parseTweet  
    51. 
  
  
  
  51.    // Build up the table of tweets indexed by user  
    52. 
  
  
  
  52.    |> Event.scan (fun z x -> addToMultiMap x.UserName x z) Map.empty  
    53. 
  
  
  
  53.    // Take every 20’ˉth index  
    54. 
  
  
  
  54.    |> every 20  
    55. 
  
  
  
  55.    // Listen and display the average of #tweets/user  
    56. 
  
  
  
  56.    |> Event.add (fun s -> 
    57. 
  
  
  
  57.         let avg = s |> Seq.averageBy (fun (KeyValue(_,d)) -> float d.Length)  
    58. 
  
  
  
  58.         printfn "#users = %d, avg tweets = %g" s.Count avg)  
    59.

twitterStream.StartListening()以上代码对采样消息流的内容进行统计，每收到20条消息便打印出每个用户的平均推数。

 
 
 
    
  
  
  
  1. #users = 19, avg tweets = 1.05263  
    2. 
  
  
  
  2. #users = 39, avg tweets = 1.02564  
    3. 
  
  
  
  3. #users = 59, avg tweets = 1.01695  
    4. 
  
  
  
  4. #users = 79, avg tweets = 1.01266  
    5. 
  
  
  
  5. #users = 99, avg tweets = 1.0101  
    6. 
  
  
  
  6. #users = 118, avg tweets = 1.01695  
    7. 
  
  
  
  7. #users = 138, avg tweets = 1.01449  
    8. 
  
  
  
  8. #users = 158, avg tweets = 1.01266  
    9. 
  
  
  
  9. #users = 178, avg tweets = 1.01124  
    10. 
  
  
  
  10. #users = 198, avg tweets = 1.0101  
    11. 
  
  
  
  11. #users = 218, avg tweets = 1.00917  
    12. 
  
  
  
  12. #users = 237, avg tweets = 1.01266  
    13. 
  
  
  
  13. #users = 257, avg tweets = 1.01167  
    14. 
  
  
  
  14. #users = 277, avg tweets = 1.01083  
    15. 
  
  
  
  15. #users = 297, avg tweets = 1.0101  
    16. 
  
  
  
  16. #users = 317, avg tweets = 1.00946  
    17. 
  
  
  
  17. #users = 337, avg tweets = 1.0089  
    18. 
  
  
  
  18. #users = 357, avg tweets = 1.0084  
    19. 
  
  
  
  19. #users = 377, avg tweets = 1.00796  
    20. 
  
  
  
  20. #users = 396, avg tweets = 1.0101  
    21. 
  
  
  
  21. #users = 416, avg tweets = 1.00962  
    22. 
  
  
  
  22. #users = 435, avg tweets = 1.01149  
    23. 
  
  
  
  23. #users = 455, avg tweets = 1.01099  
    24. 
  
  
  
  24. #users = 474, avg tweets = 1.01266  
    25. 
  
  
  
  25. #users = 494, avg tweets = 1.01215  
    26. 
  
  
  
  26. #users = 514, avg tweets = 1.01167  
    27. 
  
  
  
  27. #users = 534, avg tweets = 1.01124  
    28. 
  
  
  
  28. #users = 554, avg tweets = 1.01083  
    29. 
  
  
  
  29. #users = 574, avg tweets = 1.01045  
    30. 
  
  
  
  30. #users = 594, avg tweets = 1.0101 
    31.

只要使用稍稍不同的分析方式，我们便可以显示出Twitter提供的采样消息流中发推超过1次的用户，以及他们最新的推内容。以下代码可以在F#的交互命令行中使用，如之前文章中的做法，在数据表格中显示内容：

 
 
 
    
  
  
  
  1. open System.Drawing  
    2. 
  
  
  
  2. open System.Windows.Forms  
    3. 
  
  
  
  3.    
    4. 
  
  
  
  4. let form = new Form(Visible = true, Text = "A Simple F# Form", TopMost = true, SizeSize = Size(600,600))  
    5. 
  
  
  
  5.    
    6. 
  
  
  
  6. let data = new DataGridView(Dock = DockStyle.Fill, Text = "F# Programming is Fun!",  
    7. 
  
  
  
  7. Font = new Font("Lucida Console",12.0f),  
    8. 
  
  
  
  8. ForeColor = Color.DarkBlue)  
    9. 
  
  
  
  9.    
    10. 
  
  
  
  10. form.Controls.Add(data)  
    11. 
  
  
  
  11.    
    12. 
  
  
  
  12. data.DataSource <- [| (10,10,10) |]  
    13. 
  
  
  
  13.    
    14. 
  
  
  
  14. data.Columns.[0].Width <- 200  
    15. 
  
  
  
  15. data.Columns.[2].Width <- 500  
    16. 
  
  
  
  16.    
    17. 
  
  
  
  17. twitterStream.NewTweet  
    18. 
  
  
  
  18.    |> Event.choose parseTweet  
    19. 
  
  
  
  19.    // Build up the table of tweets indexed by user  
    20. 
  
  
  
  20.    |> Event.scan (fun z x -> addToMultiMap x.UserName x z) Map.empty  
    21. 
  
  
  
  21.    // Take every 20’ˉth index  
    22. 
  
  
  
  22.    |> every 20  
    23. 
  
  
  
  23.    // Listen and display those with more than one tweet  
    24. 
  
  
  
  24.    |> Event.add (fun s -> 
    25. 
  
  
  
  25.         let moreThanOneMessage = s |> Seq.filter (fun (KeyValue(_,d)) -> d.Length > 1)   
    26. 
  
  
  
  26.         data.DataSource <-   
    27. 
  
  
  
  27.             moreThanOneMessage  
    28. 
  
  
  
  28.             |> Seq.map (fun (KeyValue(user,d)) -> (user, d.Length, d.Head.Status))  
    29. 
  
  
  
  29.             |> Seq.filter (fun (_,n,_) -> n > 1)  
    30. 
  
  
  
  30.             |> Seq.sortBy (fun (_,n,_) -> -n)  
    31. 
  
  
  
  31.             |> Seq.toArray) 
    32.

twitterStream.StartListening()以下是部分采样结果：请注意，在上面的示例中，我们使用阻塞式的I/O操作来读取Twitter消息流。这有两个原因──Twitter数据流十分活跃(且一直如此)，而且我们可以假设不会有太多的Twitter流──如这里只有1个。此外，Twitter会对单一帐号的采样次数进行限制。文章后续的内容中，我们会演示如何对此类XML片段进行非阻塞的读取。

用F#做并行，用C#/VB做GUI，反馈进度的事件模式，对于那种F#程序员实现异步计算组件，并交给C#或VB程序员来使用的场景非常有用。在下面的示例中，发布出去的事件需要被标记为[]，以此保证它们在C#或VB程序员看来也是标准的事件。例如在上面第二个示例中，您需要使用：

 
 
 
    
  
  
  
  1. /// Raised when a particular job completes  
    2. 
  
  
  
  2. []  
    3. 
  
  
  
  3. member x.JobCompleted  = jobCompleted.Publish  
    4. 
  
  
  
  4.  
    5. 
  
  
  
  5. /// Raised when all jobs complete  
    6. 
  
  
  
  6. []  
    7. 
  
  
  
  7. member x.AllCompleted  = allCompleted.Publish  
    8. 
  
  
  
  8.  
    9. 
  
  
  
  9. /// Raised when the composition is cancelled successfully  
    10. 
  
  
  
  10. []  
    11. 
  
  
  
  11. member x.Canceled   = canceled.Publish  
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13. /// Raised when the composition exhibits an error  
    14. 
  
  
  
  14. []  
    15. 
  
  
  
  15. member x.Error      = error.Publish模式的限制  
    16.

反馈进度的事件模式会有一些假设：并行处理组件的使用者是那些GUI应用程序（如Windows Forms），服务器端应用程序（如ASP.NET）或其他一些能够将事件交由监控方使用场景。我们也可以调整这一模式中发起事件的方式，例如将消息发送给一个MailboxProcessor或简单地记录它们。然而这里还是有一些假设，需要有个主线程或是其他某个监控者来监听这些事件，或是合理的保存它们。

反馈进度的事件模式同样假设封装后对象可以获取GUI线程的同步上下文，这通常是隐式的（如上面那些例子）。这一般是个合理的假设。还有一种做法是由外部参数来获得这个上下文，虽然它在.NET编程中并非是种常见的做法。

如果您对于.NET 4.0中的IObservable接口较为熟悉，您可能会考虑让TwitterStreamSample类型实现这个接口。然而，对于最终数据源来说，这个做法的好处不大。例如，以后TwitterStreamSample类型可能会需要提供更多种事件，例如在发生错误并自动重建连接时汇报，或是汇报暂停或延迟状况。在这样的场景中，发起.NET事件就够了，部分原因是为了让更多.NET程序员熟悉这个对象。在F#种，所有发布出去的IEvent<_>对象会自动实现IObservable，这样其他人在使用时便可以直接使用Observable组合器。

结论

反馈进度的事件模式是一种用于强大而优雅的做法，用于在某个边界之后对并行的执行过程加以封装，并同时汇报执行的结果或是进度。在外部，AsyncWoker对象的表现形式一般是单线程的。假设您的异步输入是独立的，这意味着该组件不需要将程序的其他部分暴露在多线程的竞争条件下面。

所有的JavaScript，ASP.NET以及GUI框架的程序员（如Windows Forms）都明白，框架的单线程特性既是优势也是劣势──问题变得简单了（没有数据竞争），但并行和异步编程却变得很困难。在.NET编程中，I/O和繁重的CPU计算必须交由后台线程去处理。上面的设计模式可以同时给您两个世界的优势：您得到了独立的，可以互操作的，通信丰富的后台处理组件，其中包括了对I/O及并行计算的支持，同时还在您的大部分代码中保留了单线程GUI编程的简单性。正如之前表现的那样，这些组件还保持了很高的通用性及可复用性，这使得独立的单元测试也变得非常容易。

文章转自老赵的博客，

原文链接：http://blog.zhaojie.me/2010/03/async-and-parallel-design-patterns-in-fsharp-2-reporting-progress-with-events.html

【编辑推荐】

  1. 详解F#对象序列化为XML的实现方法
  2. F#运算符定义规则总结
  3. 浅析F#简易Comet聊天服务实例
  4. 详解F#版本的CodeTimer方法实现
  5. TechED 09视频专访：F#与函数式编程语言

本文题目：F#中的异步及并行模式：反馈进度的事件
标题路径：http://www.stwzsj.com/qtweb/news33/2633.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站内链知识

行业网站建设

成都展柜厂　　 　上市集团网站建设　　 　商城网站建设公司　　 　成都400电话申请　　 　成都app软件开发　　 　南充网站建设　　 　成都枢纽中心　　 　响应式网站　　 　温江网站制作　　 　成都网站设计　　 　高端网站策划　　 　一带一路　　 　成都响应式网站建设公司　　 　达州主机托管　　 　四川黔盛装饰　　 　胜锦安防科技　　 　成都印刷厂　　 　成都企业网站设计　　 　成都茶叶销售　　 　自贡托管服务器