ScalaActor：多线程的基础学习

Scala Actor是Scala里多线程的基础，核心思想是用消息传递来进行线程间的信息共享和同步。

河曲ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为创新互联建站的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：18982081108（备注：SSL证书合作）期待与您的合作！

编辑推荐：Scala编程语言专题

Scala Actor线程模型可以这样理解：所有Actor共享一个线程池，总的线程个数可以配置，也可以根据CPU个数决定；当一个Actor启动之后，Scala分配一个线程给它使用，如果使用receive模型，这个线程就一直为该Actor所有，如果使用react模型，Scala执行完react方法后抛出异常，则该线程就可以被其它Actor使用。

下面看一些核心代码。

 
 
 
    
  
  
  
  1.  def start(): Actor = synchronized {  
    2.   
  
  
  2.   // Reset various flags.  
    3.   
  
  
  3.   //  
    4.   
  
  
  4.   // Note that we do *not* reset `trapExit`. The reason is that  
    5.   
  
  
  5.   // users should be able to set the field in the constructor  
    6.   
  
  
  6.   // and before `act` is called.  
    7.   
  
  
  7.  
    8.   
  
  
  8.   exitReason = 'normal  
    9.   
  
  
  9.   exiting = false 
    10.   
  
  
  10.   shouldExit = false 
    11.   
  
  
  11.  
    12.   
  
  
  12.   scheduler execute {  
    13.   
  
  
  13.     ActorGC.newActor(Actor.this)  
    14.   
  
  
  14.     (new Reaction(Actor.this)).run()  
    15.   
  
  
  15.   }  
    16.   
  
  
  16.  
    17.   
  
  
  17.   this 
    18.   
  
  
  18. }  
    19.

其中Reaction实现Runnable接口，scheduler基本相当于是一个线程池，所以调用start方法之后会有一个线程来为该Actor服务。

使用receive模型。

 
 
 
    
  
  
  
  1. def receive[R](f: PartialFunction[Any, R]): R = {  
    2.   
  
  
  2.  assert(Actor.self == this, "receive from channel belonging to other actor")  
    3.   
  
  
  3.  this.synchronized {  
    4.   
  
  
  4.    if (shouldExit) exit() // links  
    5.   
  
  
  5.    val qel = mailbox.extractFirst((m: Any) => f.isDefinedAt(m))  
    6.   
  
  
  6.    if (null eq qel) {  
    7.   
  
  
  7.      waitingFor = f.isDefinedAt  
    8.   
  
  
  8.      isSuspended = true 
    9.   
  
  
  9.      suspendActor()  
    10.   
  
  
  10.    } else {  
    11.   
  
  
  11.      received = Some(qel.msg)  
    12.   
  
  
  12.      sessions = qel.session :: sessions  
    13.   
  
  
  13.    }  
    14.   
  
  
  14.    waitingFor = waitingForNone  
    15.   
  
  
  15.    isSuspended = false 
    16.   
  
  
  16.  }  
    17.   
  
  
  17.  val result = f(received.get)  
    18.   
  
  
  18.  sessions = sessions.tail  
    19.   
  
  
  19.  result  
    20.   
  
  
  20.   
    21.

如果当前mailbox里面没有可以处理的消息，调用suspendActor，该方法会调用wait；如果有消息，这调用PartialFunction进行处理。

使用react模型。

 
 
 
    
  
  
  
  1. def react(f: PartialFunction[Any, Unit]): Nothing = {  
    2.   
  
  
  2.  assert(Actor.self == this, "react on channel belonging to other actor")  
    3.   
  
  
  3.  this.synchronized {  
    4.   
  
  
  4.    if (shouldExit) exit() // links  
    5.   
  
  
  5.    val qel = mailbox.extractFirst((m: Any) => f.isDefinedAt(m))  
    6.   
  
  
  6.    if (null eq qel) {  
    7.   
  
  
  7.      waitingFor = f.isDefinedAt  
    8.   
  
  
  8.      continuation = f  
    9.   
  
  
  9.      isDetached = true 
    10.   
  
  
  10.    } else {  
    11.   
  
  
  11.      sessions = List(qel.session)  
    12.   
  
  
  12.      scheduleActor(f, qel.msg)  
    13.   
  
  
  13.    }  
    14.   
  
  
  14.    throw new SuspendActorException  
    15.   
  
  
  15.  }  
    16.   
  
  
  16.   
    17.

如果当前mailbox没有可以处理的消息，设置waitingFor和continuation，这两个变量会在接收到消息的时候使用；如果有消息，则调用scheduleActor，该方法会在线程池里选择一个新的线程来处理，具体的处理方法也是由PartialFunction决定。不管是哪条路径，react都会立即返回，或者说是立即抛出异常，结束该线程的执行，这样该线程就可以被其它Actor使用。

再来看看接收消息的处理代码。

 
 
 
    
  
  
  
  1. def send(msg: Any, replyTo: OutputChannel[Any]) = synchronized {  
    2.   
  
  
  2.  if (waitingFor(msg)) {  
    3.   
  
  
  3.    received = Some(msg)  
    4.   
  
  
  4.  
    5.   
  
  
  5.    if (isSuspended)  
    6.   
  
  
  6.      sessions = replyTo :: sessions  
    7.   
  
  
  7.    else 
    8.   
  
  
  8.      sessions = List(replyTo)  
    9.   
  
  
  9.  
    10.   
  
  
  10.    waitingFor = waitingForNone  
    11.   
  
  
  11.  
    12.   
  
  
  12.    if (!onTimeout.isEmpty) {  
    13.   
  
  
  13.      onTimeout.get.cancel()  
    14.   
  
  
  14.      onTimeout = None  
    15.   
  
  
  15.    }  
    16.   
  
  
  16.  
    17.   
  
  
  17.    if (isSuspended)  
    18.   
  
  
  18.      resumeActor()  
    19.   
  
  
  19.    else // assert continuation != null  
    20.   
  
  
  20.      scheduler.execute(new Reaction(this, continuation, msg))  
    21.   
  
  
  21.  } else {  
    22.   
  
  
  22.    mailbox.append(msg, replyTo)  
    23.   
  
  
  23.  }   
    24.

如果当前没有在等待消息或者接收到的消息不能处理，就丢到mailbox里去；相反，则进行消息的处理。这里对于receive模型和react模型就有了分支：如果isSuspended为true，表示是receive模型，并且线程在wait，就调用resumeActor，该方法会调用notify；否则就是react模型，同样在线程池里选择一个线程进行处理。

这样，相信大家对Scala Actor就有了一个基本的认识。

【相关阅读】

  1. Scala入门介绍：Hello World
  2. Scala初学者学习资料：main(String[])
  3. 影响Scala语言设计的因素列表
  4. 喜欢Scala编程的四个理由
  5. Scala融合面向对象和函数概念的方法

网站名称：ScalaActor：多线程的基础学习
当前网址：http://www.gawzjz.com/qtweb/news13/188513.html

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

广告

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

猜你还喜欢下面的内容

自适应网站知识

同城分类信息

成都移动机房　　 　德阳东方电机技改　　 　双流能亿建站　　 　成都LED显示屏　　 　做手机网站　　 　高端网站设计　　 　云服务器　　 　教育建设方案　　 　网站排名　　 　成都门窗定制　　 　全网营销　　 　微信小程序开发　　 　光华机房服务器托管　　 　重庆网站制作　　 　成都营销型网站建设　　 　达州托管服务器　　 　绵阳科技城电信IDC机房　　 　企业画册设计制作　　 　内江做网站　　 　移动手机网站制作