详解F#异步及并行模式中的轻量级代理

F#目前还有些待字闺中的意思，不过随着大家对F#了解的加深。希望更多的程序员能运用好F#。在此之前，我们曾报道过《详解F#异步及并行模式中的并行CPU及I/O计算》

创新互联自2013年创立以来，是专业互联网技术服务公司，拥有项目成都网站建设、成都网站设计网站策划，项目实施与项目整合能力。我们以让每一个梦想脱颖而出为使命，1280元扶余做网站,已为上家服务,为扶余各地企业和个人服务,联系电话:028-86922220

在本系列的第3部分中，我们会来探索F#中轻量级的，交互式的代理，以及与代理有关的一些模式，包括“隔离的内部状态”。（译注：由于原文较长，因此译文分为两段，目前是第一段，讲解了F#中异步代理的基本使用方式。）

第1部分描述了F#作为一个并行及异步语言，是如何支持轻量级的响应操作，并给出了CPU异步并行和I/O异步并行两种模式。

第2部分描述了如何从异步计算或后台计算单元中获得结果。

模式4：您的第一个代理

我们来观察您所创建的第一个异步代理：

 
 
 
  
  
  type Agent<'T> = MailboxProcessor<'T>
  
  
  let agent = 
  
  
     Agent.Start(fun inbox ->
  
  
       async { while true do 
  
  
                 let! msg = inbox.Receive() 
  
  
                 printfn "got message '%s'" msg } )

这个代理不断地异步等待消息，并将它们打印出来。在这段代码中，每个消息都是一个字符串，且agent的类型是：

agent.Post "hello!"这便会打印出：

got message 'hello!'也可以这样发送多条消息：

for i in 1 .. 10000 do
agent.Post (sprintf "message %d" i)

这样便可以打印出10000条消息。

您可以认为每个代理对象都包含一个消息队列（或管道），并在消息到达时进行响应。一个委托一般都使用异步的循环等待来消息并进行处理。如在上面的例子中，代理使用while循环进行处理。

许多读者可能已经对代理颇为熟悉了。如Erlang，它便是基于代理设计的（在那里被称为进程）。而不久之前，一个基于.NET平台的实验性的孵化型语言，Axum，也注重了基于代理编程的重要性。Axum与F#中的代理设计相互影响，而其他包含轻量级线程的语言也强调了基于代理的组合与设计。

上面的例子一开始创建了一个类型的缩写：Agent，它代表了F#类库中基于内存的代理类型“MailboxProcessor”。如果您愿意的话也可以使用这个完整的名字，不过我更喜欢简单的命名。

您的第一批10万个代理

代理对象非常轻量，这是因为它基于F#的异步编程模型。例如，您可以在一个.NET进程中创建成百上千，甚至更多个代理。例如，我们来创建10万个简单的代理对象：

 
 
 
  
  
  let agents = 
  
  
      [ for i in 0 .. 100000 ->
  
  
         Agent.Start(fun inbox ->
  
  
           async { while true do 
  
  
                     let! msg = inbox.Receive() 
  
  
                     if i % 10000 = 0 then 
  
  
                         printfn "agent %d got message '%s'" i msg } ) ]

您可以这样向每个代理对象发送消息：

for agent in agents do
agent.Post "ping!每第1万个代理对象会在收到消息时打印信息。这个代理集合在处理消息时非常迅速，只要几秒钟时间。代理和内存中的消息处理非常快。

很显然，代理并不与.NET线程直接对应──您不可能在单个应用程序中创建10万的线程（在32位操作系统中，即便1000个线程也已经太多了）。相反，在代理等待消息时，它实际上只是表现为一个回调函数，一些对象分配，以及代理所引用的闭包等等。在收到消息之后，代理的工作会在一个线程池（默认便是.NET线程池）中分配并执行。

尽管需要10万个代理的情况并不多见，不过2000多个代理倒是很正常的。接下来我们便会看到这样一些例子。

高伸缩的Web服务器处理请求

在F#编程中，异步代理的思想其实是一种在多个环境中反复出现的设计模式。在F#中，我们经常使用“代理”这个词表示一种随时发生的，特别是通过循环，或是处理消息，或是产生结果的异步计算。

例如，在以后的文章中，我们会来关注如何使用F#构建伸缩性强的TCP或HTTP服务器应用程序，并将它们部署到EC2或是Windows Azure中去。这里我们打算用“股票服务器”作为例子，它接受TCP或HTTP连接，并向客户端返回一系列的股票信息。每个客户端会每隔一秒钟收到一条股票信息。这个服务最终会以单个URL或REST API的形式发布。

在实现时，我们为每个客户端请求分配一个异步代理（由于只是演示，我们在这里便不断地写入相同的AAPL股票信息）：

 
 
 
  
  
  open System.Net.Sockets 
  
  
  /// serve up a stream of quotes 
  
  
  let serveQuoteStream (client: TcpClient) = async { 
  
  
      let stream = client.GetStream() 
  
  
      while true do
  
  
          do! stream.AsyncWrite( "AAPL 200.38"B ) 
  
  
          do! Async.Sleep 1000.0 // sleep one second}

每个代理会一直运行到客户端连接断开。因为代理非常轻量，因此这个股票服务能够在一台机器上支持数千个并发连接（如果使用云托管服务则会有更好的伸缩性）。而同一时刻会出现多少个代理对象则取决于客户端的数量。

上面的例子演示了使用F#进行网络编程是多么的方便──网络协议在此变成了基于异步代理的数据流读写。在以后的文章中我们会观察更多使用F#进行伸缩性强的TCP/HTTP编程。

代理与隔离状态（命令式）

F#代理编程的一个优秀的关键之处便是其隔离性。隔离性则意味着资源“归属”与某个特定的代理，而不会暴露给其他代理。因此，独立状态对并发的访问及数据竞争是一种良好的保护。

在F#中，异步代理的独立性直接表现为文法上的作用域。例如，下面的代码使用一个字典来累计发送至代理对象的不同消息的次数。内部的字典在文法上是异步代理私有的，因此我们无法在代理外部对字典进行读写。这意味着字典的可变状态实际上是被隔离的，代理保证了对它的非并发的安全访问。

 
 
 
  
  
  type Agent<'T> = MailboxProcessor<'T>
  
  
  open System.Collections.Generic 
  
  
  let agent = 
  
  
     Agent.Start(fun inbox ->
  
  
       async { let strings = Dictionary() 
  
  
               while true do 
  
  
                 let! msg = inbox.Receive() 
  
  
                 if strings.ContainsKey msg then 
  
  
                     strings.[msg] <- strings.[msg] + 1 
  
  
                 else 
  
  
                     strings.[msg] <- 0 
  
  
                 printfn "message '%s' now seen '%d' times" msg strings.[msg] } )

状态隔离是F#的基本特性，它并不是代理所独有的。例如，下面的代码对StreamReader和ResizeArray（这是F#中对System.Collections.Generics.List的别称）的隔离访问。请注意这段代码和.NET类库中的System.IO.File.ReadAllLines方法功能相同：

 
 
 
  
  
  let readAllLines (file:string) = 
  
  
      use reader = new System.IO.StreamReader(file) 
  
  
      let lines = ResizeArray<_>() 
  
  
      while not reader.EndOfStream do 
  
  
          lines.Add (reader.ReadLine()) 
  
  
      lines.ToArray()

在这里，reader和ResizeArray对象都无法在函数外部使用。在代理或其他持续计算的情况里，隔离性是至关重要的──状态在这里永远独立于程序运行中的其他部分。

说到底，隔离性是个动态的属性，它经常受到文法的约束。例如，考虑这样一个延迟的，随需加载的读取器，它会读取文件中的所有行：

 
 
 
  
  
  let readAllLines (file:string) = 
  
  
      seq { use reader = new System.IO.StreamReader(file) 
  
  
            while not reader.EndOfStream do
  
  
                yield reader.ReadLine() }

隔离状态经常包含可变的值，包括F#中的引用单元。例如，下面的代码在一个引用单元中不断累计接受到的消息个数：

 
 
 
  
  
  let agent = 
  
  
     Agent.Start(fun inbox -> 
  
  
       async { let count = ref 0 
  
  
               while true do
  
  
                 let! msg = inbox.Receive() 
  
  
                 incr count 
  
  
                 printfn "now seen a total of '%d' messages" !count } )

再次强调，这里可变的状态被隔离了，确保了对它单线程的安全访问。

在代理中使用循环和隔离状态（函数式）

F#程序员了解两种实现循环的方法：使用“命令式”的while/for以及可变的累加器（ref或mutable），或是“函数式”风格的调用，将累加状态作为参数传递给一个或多个递归函数。例如，计算文件行数的程序可以使用命令式的方式来写：

 
 
 
  
  
  let countLines (file:string) = 
  
  
      use reader = new System.IO.StreamReader(file) 
  
  
      let count = ref 0 
  
  
      while not reader.EndOfStream do
  
  
          reader.ReadLine() |> ignore 
  
  
          incr count 
  
  
      !count或是递归式的： 
  
  
  
  
  
  let countLines (file:string) = 
  
  
      use reader = new System.IO.StreamReader(file) 
  
  
      let rec loop n = 
  
  
          if reader.EndOfStream then n 
  
  
          else
  
  
              reader.ReadLine() |> ignore 
  
  
              loop (n+1) 
  
  
      loop 0

在使用时，两种方式都是可行的：函数式的做法相对更加高级一些，但是大大减少了代码中显式的状态修改次数，且更为通用。两种写法对于F#程序员来说，一般都可以理解，他们还可以将“while”循环转化为等价的“let rec”函数（这是个不错的面试问题！）。

有趣的是，在编写异步循环时的规则也一样：您可以使用命令式的“while”或函数式的“let rec”中的任意一种来定义循环。例如，这里有一个利用递归的异步函数统计消息数量的做法：

 
 
 
  
  
  let agent = 
  
  
     Agent.Start(fun inbox -> 
  
  
       let rec loop n = async { 
  
  
           let! msg = inbox.Receive() 
  
  
           printfn "now seen a total of %d messages" (n+1) 
  
  
           return! loop (n+1) 
  
  
       } 
  
  
       loop 0 )

这样我们便可以获得这样的输出：

now seen a total of 0 messages
now seen a total of 1 messages
....
now seen a total of 10000 messages

再提一次，定义代理对象的两种常见模式为命令式的：

 
 
 
  
  
  let agent = 
  
  
     Agent.Start(fun inbox ->
  
  
       async { 
  
  
           // isolated imperative state goes here 
  
  
           ... 
  
  
           while  do 
  
  
               // read messages and respond 
  
  
               ... 
  
  
       })

及函数式的：

 
 
 
  
  
  let agent =  
  
  
      Agent.Start(fun inbox ->
  
  
        let rec loop arg1 arg2 = async {  
  
  
            // receive and process messages here 
  
  
            ... 
  
  
            return! loop newArg1 newArg2 
  
  
         } 
  
  
  
  
  
        loop initialArg1 initialArg2)

再次强调，两种方法在F#都是合理的──使用递归的异步函数可能是更高级的方法，但更为函数式且更为通用。

链接：http://www.cnblogs.com/JeffreyZhao/archive/2010/03/15/async-and-parallel-design-patterns-in-fsharp-3-agents.html

【F#教程回顾】

F#简明教程一：F#与函数式编程概述
F#简明教程二：F#类型系统和类型推断机制
F#简明教程三：F#语法精要
详解F#对象序列化为XML的实现方法
F#运算符定义规则总结

网站栏目：详解F#异步及并行模式中的轻量级代理
当前网址：http://www.mswzjz.cn/qtweb/news46/534796.html

攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等

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

贝锐智能技术为您推荐以下文章