现在的位置: 首页 > 综合 > 正文

scala学习十二 并发编程二 actor模型

2018年03月21日 ⁄ 综合 ⁄ 共 9172字 ⁄ 字号 评论关闭

在 Scala 中实现并发编程的主要方法:可以直接使用 Java 语言的并发库的基本方法,可以使用 Scala API 中的MailBox 类型。尽管这两种方法都是可行的,但是它们并不是
Scala 实现并发性的主要机制。真正提供并发性的是 Scala 的 actor

什么是 “actor”?

“actor” 实现在称为 actor 的执行实体之间使用消息传递进行协作(注意,这里有意避免使用 “进程”、“线程” 或 “机器” 等词汇)。尽管它听起来与
RPC 机制有点儿相似,但是它们是有区别的。RPC 调用(比如 Java RMI 调用)会在调用者端阻塞,直到服务器端完成处理并发送回某种响应(返回值或异常),
而消息传递方法不会阻塞调用者,因此可以巧妙地避免死锁

仅仅传递消息并不能避免错误的并发代码的所有问题。另外,这种方法还有助于使用 “不共享任何东西” 编程风格,也就是说不同的 actor 并不访问共享的数据结构(这有助于促进封装 actor,无论 actor 是 JVM 本地的,还是位于其他地方) — 这样就完全不需要同步了。毕竟,如果不共享任何东西,并发执行就不涉及任何需要同步的东西。

这不算是对 actor 模型的正规描述,而且毫无疑问,具有更正规的计算机科学背景的人会找到各种更严谨的描述方法,能够描述 actor 的所有细节。但是对于本文来说,这个描述已经够了。在网上可以找到更详细更正规的描述,还有一些学术文章详细讨论了 actor 背后的概念(请您自己决定是否要深入学习这些概念)。现在,我们来看看 Scala actors API。

使用 actor 根本不困难,只需使用 Actor 类的 actor 方法创建一个
actor

import scala.actors._, Actor._

package com.tedneward.scalaexamples.scala.V4
{
  object Actor1
  {
    def main(args : Array[String]) =
    {
      val badActor =
        actor//匿名函数
        {
          receive//函数
          {
            case msg => System.out.println(msg)
          }
        }
      badActor ! "Do ya feel lucky, punk?"
    }
  }
}

如果上面代码有什么疑问,大部分都是因为scala的简写:

上面代码如下:

object actor11 {
  def main(args: Array[String]) =
    {
      val badActor =
        actor {
          receive {
            (msg) =>
              msg match {
                case some => System.out.println(msg)
              }
          }
        }
      println("还么有执行消息呦");
      badActor ! "Do ya feel lucky, punk?"
    }
}


这里同时做了两件事。

首先,我们从 Scala Actors 库的包中导入了这个库,然后从库中直接导入了 Actor 类的成员;第二步并不是完全必要的,因为在后面的代码中可以使用 Actor.actor 替代 actor,但是这么做能够表明 actor 是语言的内置结构并(在一定程度上)提高代码的可读性。

下一步是使用 actor 方法创建
actor 本身,这个方法通过参数接收一个代码块。在这里,代码块执行一个简单的 receive(稍后讨论)。结果是一个
actor,它被存储在一个值引用中,供以后使用。

请记住,除了消息之外,actor 不使用其他通信方法。使用 ! 的代码行实际上是一个向 badActor 发送消息的方法,这可能不太直观。Actor 内部还包含另一个 MailBox 元素(已讨论);! 方法接收传递过来的参数(在这里是一个字符串),把它发送给邮箱,然后立即返回。

消息交付给 actor 之后,actor 通过调用它的 receive 方法来处理消息;这个方法从邮箱中取出第一个可用的消息,把它交付给一个模式匹配块。注意,因为这里没有指定模式匹配的类型,所以任何消息都是匹配的,而且消息被绑定到 msg 名称(为了打印它)。

一定要注意一点:对于可以发送的类型,没有任何限制 — 不一定要像前面的示例那样发送字符串。实际上,基于 actor 的设计常常使用 Scala case 类携带实际消息本身,这样就可以根据 case 类的参数/成员的类型提供隐式的 “命令” 或 “动作”,或者向动作提供数据。

例如,假设希望 actor 用两个不同的动作来响应发送的消息;代码如下(模式匹配):

object Actor2 {
  case class Speak(line: String);
  case class Gesture(bodyPart: String, action: String);
  case class NegotiateNewContract;

  def main(args: Array[String]) =
    {
      import scala.actors._, Actor._
      val badActor =
        actor {
          receive {
          case NegotiateNewContract =>
              System.out.println("I won't do it for less than $1 million!")
              
            case Speak(line) =>
              System.out.println(line)
              
            case Gesture(bodyPart, action) =>
              System.out.println("(" + action + "s " + bodyPart + ")")
              System.out.println("Huh? I'll be in my trailer.")
          }
        }

      badActor ! NegotiateNewContract
      badActor ! Speak("Do ya feel lucky, punk?")
      badActor ! Gesture("face", "grimaces")
      badActor ! Speak("Well, do ya?")
    }
}</span></span>

上面的代码虽然运用了模式匹配可以进行不同的动作,但是在程序运行期间只会接受一个消息,所以要想程序循环的接收消息需要将其放在循环中:

object Actor2
  {
    case class Speak(line : String);
    case class Gesture(bodyPart : String, action : String);
    case class NegotiateNewContract;
    case class ThatsAWrap;
  
    def main(args : Array[String]) =
    {
      val badActor =
        actor
        {
<span style="white-space:pre">	</span> var done = false
<span style="white-space:pre">	</span> while (! done)
          {
            receive
            {
              case NegotiateNewContract =>
                System.out.println("I won't do it for less than $1 million!")
              case Speak(line) =>
                System.out.println(line)
              case Gesture(bodyPart, action) =>
                System.out.println("(" + action + "s " + bodyPart + ")")
              case ThatsAWrap =>
                System.out.println("Great cast party, everybody! See ya!")
                done = true
              case _ =>
                System.out.println("Huh? I'll be in my trailer.")
            }
          }
        }
      
      badActor ! NegotiateNewContract
      badActor ! Speak("Do ya feel lucky, punk?")
      badActor ! Gesture("face", "grimaces")
      badActor ! Speak("Well, do ya?")
      badActor ! ThatsAWrap
    }
  }</span>

通过这样的运用,感觉像是异步调用,但这其实不是的:

object Actor3
  {
    case class Speak(line : String);
    case class Gesture(bodyPart : String, action : String);
    case class NegotiateNewContract;
    case class ThatsAWrap;
  
    def main(args : Array[String]) =
    {
      def ct =
        "Thread " + Thread.currentThread().getName() + ": "
      val badActor =
        actor
        {
          var done = false
          while (! done)
          {
            receive
            {
              case NegotiateNewContract =>
                System.out.println(ct + "I won't do it for less than $1 million!")
              case Speak(line) =>
                System.out.println(ct + line)
              case Gesture(bodyPart, action) =>
                System.out.println(ct + "(" + action + "s " + bodyPart + ")")
              case ThatsAWrap =>
                System.out.println(ct + "Great cast party, everybody! See ya!")
                done = true
              case _ =>
                System.out.println(ct + "Huh? I'll be in my trailer.")
            }
          }
        }
      
      System.out.println(ct + "Negotiating...")
      badActor ! NegotiateNewContract
      System.out.println(ct + "Speaking...")
      badActor ! Speak("Do ya feel lucky, punk?")
      System.out.println(ct + "Gesturing...")
      badActor ! Gesture("face", "grimaces")
      System.out.println(ct + "Speaking again...")
      badActor ! Speak("Well, do ya?")
      System.out.println(ct + "Wrapping up")
      badActor ! ThatsAWrap
    }
  }</span>

运行结果:

<span style="font-size:14px;">Thread main: Negotiating...
Thread main: Speaking...
Thread main: Gesturing...
Thread ForkJoinPool-1-worker-13: I won't do it for less than $1 million!
Thread ForkJoinPool-1-worker-13: Do ya feel lucky, punk?
Thread <span style="color:#ff0000;">main</span>: Speaking again...
Thread main: Wrapping up
Thread <span style="color:#ff0000;">ForkJoinPool</span>-1-worker-13: (grimacess face)
Thread ForkJoinPool-1-worker-13: Well, do ya?
Thread ForkJoinPool-1-worker-13: Great cast party, everybody! See ya!
</span>

存在一个main线程和一个ForkJoinPool线程(actor框架幕后生成的)

回顾一下上一篇中的生产者与消费者,在看看下面通过actor模型完成的例子:

<span style="font-size:14px;">  object ProdConSample1
  {
    case class Message(msg : String)
    
    def main(args : Array[String]) : Unit =
    {
      val consumer =
        actor
        {
          var done = false
          while (! done)
          {
            receive
            {
              case msg =>
                System.out.println("Received message! -> " + msg)
                done = (msg == "DONE")
            }
          }
        }
      
      consumer ! "Mares eat oats"
      consumer ! "Does eat oats"
      consumer ! "Little lambs eat ivy"
      consumer ! "Kids eat ivy too"
      consumer ! "DONE"      
    }
  }</span>

上面的这个版本确实简短多了,而且在某些情况下可能能够完成所需的所有工作;但是,如果运行这段代码并与以前的版本做比较,就会发现一个重要的差异 — 基于 actor 的版本是一个多位置缓冲区,而不是我们以前使用的单位置缓冲。这看起来是一项改进,而不是缺陷,但是我们要通过对比确认这一点。我们来创建 Drop 的基于
actor 的版本,在这个版本中所有对 put() 的调用必须由对 take() 的调用进行平衡。

幸运的是,Scala Actors 库很容易模拟这种功能。希望让 Producer 一直阻塞,直到 Consumer 接收了消息;实现的方法很简单:让 Producer 一直阻塞,直到它从 Consumer 收到已经接收消息的确认。从某种意义上说,这就是以前的基于监视器的代码所做的,那个版本通过对锁对象使用监视器发送这种信号,
Scala Actors 库中,最容易的实现方法是使用 
!? 方法而不是 ! 方法(这样就会一直阻塞到收到确认时)。
(在
Scala Actors 实现中,每个 Java 线程都是一个 actor,所以回复会发送到与 
main 线程隐式关联的邮箱)。这意味着
Consumer 需要发送某种确认;这要使用隐式继承的
reply(它还继承 receive 方法)代码如下:

 object ProdConSample2
  {
    case class Message(msg : String)
    
    def main(args : Array[String]) : Unit =
    {
      val consumer =
        actor
        {
          var done = false
          while (! done)
          {
            receive
            {
              case msg =>
                System.out.println("Received message! -> " + msg)
                done = (msg == "DONE")
                reply("RECEIVED")
            }
          }
        }
      
      System.out.println("Sending....")
      consumer !? "Mares eat oats"
      System.out.println("Sending....")
      consumer !? "Does eat oats"
      System.out.println("Sending....")
      consumer !? "Little lambs eat ivy"
      System.out.println("Sending....")
      consumer !? "Kids eat ivy too"
      System.out.println("Sending....")
      consumer !? "DONE"      
    }
  }

如果喜欢使用 spawn 把
Producer 放在 
main() 之外的另一个线程中(这非常接近最初的代码)

object ProdConSampleUsingSpawn
  {
    import concurrent.ops._
  
    def main(args : Array[String]) : Unit =
    {
      // Spawn Consumer
      val consumer =
        actor
        {
          var done = false
          while (! done)
          {
            receive
            {
              case msg =>
                System.out.println("MESSAGE RECEIVED: " + msg)
                done = (msg == "DONE")
                reply("RECEIVED")
            }
          }
        }
    
      // Spawn Producer
      spawn
      {
        val importantInfo : Array[String] = Array(
          "Mares eat oats",
          "Does eat oats",
          "Little lambs eat ivy",
          "A kid will eat ivy too",
          "DONE"
        );
        
        importantInfo.foreach((msg) => consumer !? msg)
      }
    }
  }

无论从哪个角度来看,基于 actor 的版本都比原来的版本简单多了。读者只要让 actor 和隐含的邮箱自己发挥作用即可。

但是,这并不简单。actor 模型完全颠覆了考虑并发性和线程安全的整个过程;在以前的模型中,我们主要关注共享的数据结构(数据并发性),而现在主要关注操作数据的代码本身的结构(任务并发性),尽可能少共享数据。请注意
Producer/Consumer 示例的不同版本的差异。在以前的示例中,并发功能是围绕 Drop 类(有界限的缓冲区)显式编写的。在本文中的版本中,Drop 甚至没有出现,重点在于两个
actor(线程)以及它们之间的交互(通过不共享任何东西的消息)。

当然,仍然可以用 actor 构建以数据为中心的并发构造;只是必须采用稍有差异的方式。请考虑一个简单的 “计数器” 对象,它使用 actor 消息传达 “increment” 和 “get” 操作

 object CountingSample
  {
    case class Incr
    case class Value(sender : Actor)
    case class Lock(sender : Actor)
    case class UnLock(value : Int)
  
    class Counter extends Actor
    {
      override def act(): Unit = loop(0)
 
      def loop(value: int): Unit = {
        receive {
          case Incr()   => loop(value + 1)
          case Value(a) => a ! value; loop(value)
          case Lock(a)  => a ! value
                           receive { case UnLock(v) => loop(v) }
          case _        => loop(value)
        }
      }
    }
    
    def main(args : Array[String]) : Unit =
    {
      val counter = new Counter
      counter.start()
      counter ! Incr()
      counter ! Incr()
      counter ! Incr()
      counter ! Value(self)
      receive { case cvalue => Console.println(cvalue) }    
      counter ! Incr()
      counter ! Incr()
      counter ! Value(self)
      receive { case cvalue => Console.println(cvalue) }    
    }
  }

当然了,也可以通过Drop包装actor,提供熟悉的api:

  object ActorDropSample
  {
    class Drop
    {
      private case class Put(x: String)
      private case object Take
      private case object Stop
  
      private val buffer =
        actor
        {
          var data = ""
          loop
          {
            react
            {
              case Put(x) if data == "" =>
                data = x; reply()
              case Take if data != "" =>
                val r = data; data = ""; reply(r)
              case Stop =>
                reply(); exit("stopped")
            }
          }
        }
  
      def put(x: String) { buffer !? Put(x) }
      def take() : String = (buffer !? Take).asInstanceOf[String]
      def stop() { buffer !? Stop }
    }
    
    def main(args : Array[String]) : Unit =
    {
      import concurrent.ops._
    
      // Create Drop
      val drop = new Drop()
      
      // Spawn Producer
      spawn
      {
        val importantInfo : Array[String] = Array(
          "Mares eat oats",
          "Does eat oats",
          "Little lambs eat ivy",
          "A kid will eat ivy too"
        );
        
        importantInfo.foreach((msg) => { drop.put(msg) })
        drop.put("DONE")
      }
      
      // Spawn Consumer
      spawn
      {
        var message = drop.take()
        while (message != "DONE")
        {
          System.out.format("MESSAGE RECEIVED: %s%n", message)
          message = drop.take()
        }
        drop.stop()
      }
    }
  }

总结:

首先,actor 的主要能力来源于消息传递风格,而不采用阻塞-调用风格,这是它的主要特点。(有意思的是,也有使用消息传递作为核心机制的面向对象语言。最知名的两个例子是 Objective-C 和 Smalltalk,还有 ThoughtWorker 的 Ola Bini 新创建的 Ioke)。如果创建直接或间接扩展Actor 的类,那么要确保对对象的所有调用都通过消息传递进行。

第二,因为可以在任何时候交付消息,而且更重要的是,在发送和接收之间可能有相当长的延迟,所以一定要确保消息携带正确地处理它们所需的所有状态。这种方式会:

  • 让代码更容易理解(因为消息携带处理所需的所有状态)。
  • 减少 actor 访问某些地方的共享状态的可能性,从而减少发生死锁或其他并发性问题的机会。

第三,actor 应该不会阻塞,您从前面的内容应该能够看出这一点。从本质上说,阻塞是导致死锁的原因;代码可能产生的阻塞越少,发生死锁的可能性就越低。

很有意思的是,如果您熟悉 Java Message Service (JMS) API,就会发现我给出的这些建议在很大程度上也适用于 JMS — 毕竟,actor 消息传递风格只是在实体之间传递消息,JMS 消息传递也是在实体之间传递消息。它们的差异在于,JMS 消息往往比较大,在层和进程级别上操作;而 actor 消息往往比较小,在对象和线程级别上操作。如果您掌握了 JMS,actor 也不难掌握。

actor 并不是解决所有并发性问题的万灵药,但是它们为应用程序或库代码的建模提供了一种新的方式,所用的构造相当简单明了。尽管它们的工作方式有时与您预期的不一样,但是一些行为正是我们所熟悉的 — 毕竟,我们在最初使用对象时也有点不习惯,只要经过努力,您也会掌握并喜欢上 actor。


抱歉!评论已关闭.