第5课：彻底精通Scala隐式转换和并发编程及Spark源码阅读（2）

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

二、并发编程

       并发编程在任意的程序开发过程中都是至关重要的，并发编程的消息通信模式加上缓存的绝妙使用。并发编程至于高效的利用CPU和硬件资源是至关重要的，是高效程序中的核心中的核心。Scala中提供了actor来做并发编程，语言级别支持，scala跟java中的Thread类似。Java中的Thread是java中极大的成功也是极大的失败，成功：更好地利用了多核并发的潜能，物理硬件。失败：java中的并发是基于共享全局变量的加锁机制，这一定会不可避免地带来死锁，状态失控。分布式系统黄金准则：一定不要有全局共享的变量，更不要有加锁。Scala中的actor尽可能地减少死锁和状态共享，每个actor内部都有自己的循环器和状态。Actor框架akka。

1）实战：

    scala> import scala.actors.Actor

    import scala.actors.Actor

    scala> class HiActor extends Actor {

     |def act() {

     | while(true){

     |  receive {

     |   case name :String => println(name)

     | }}}

     | }

   defined class HiActor

   Actor是在另外一个线程中启动。

2）最佳实践receive 接收case class :

   Spark 中Master中的部分代码：

1. override def receive: PartialFunction[Any, Unit] = {
   
2.     case ElectedLeader => {
   
3.         val (storedApps, storedDrivers, storedWorkers) = persistenceEngine.readPersistedData(rpcEnv)
   
4.         state = if (storedApps.isEmpty && storedDrivers.isEmpty && storedWorkers.isEmpty) {
   
5.             RecoveryState.ALIVE
   
6.         } else {
   
7.              RecoveryState.RECOVERING
   
8.         }
   
9.        logInfo("I have been elected leader! New state: " + state)
   
10.        if (state == RecoveryState.RECOVERING) {
    
11.            beginRecovery(storedApps, storedDrivers, storedWorkers)
    
12.            recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
    
13.            override def run(): Unit = Utils.tryLogNonFatalError { self.send(CompleteRecovery) }
    
14.            }, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
    
15.        }
    
16.     }
    
17.     case CompleteRecovery => completeRecovery()
    
18.     case RevokedLeadership => {
    
19.         logError("Leadership has been revoked -- master shutting down.")
    
20.         System.exit(0)
    
21.     }

复制代码

动手案例：

  scala> case class Basic(name: String, age:Int)

  defined class Basic

  scala> case class Worker(name: String, age:Int)

  defined class Worker

  scala> class BasicActor extends Actor{

     |      def act() {

     |       while(true){

     |       receive{

     |        case Basic(name, age) => println("basic info: " + name +":" + age)

     |        case Worker(name, age) => println("worker info:" + name +": " + age)

     |       }

     |      }

     |      }

     |     }

defined class BasicActor

scala> val ac  = new BasicActor

ac: BasicActor = BasicActor@3d3b272a

scala> ac start

warning: there were 1 feature warning(s);re-run with -feature for details

basic info: wj:45

res16: scala.actors.Actor =BasicActor@3d3b272a

scala> ac ! new Basic("wj", 45)

basic info: wj:45

scala> ac ! new Worker("wj", 75)

worker info:wj: 75

备注：receive那里是一个偏函数，使用了模式匹配

！是异步发送消息，发完之后不等待

！？ 是同步发送消息，发送后会等待回复

！！是异步消息处理返回值，未来的某个时间获得结果，不会阻塞

总结：因为 actor 编程需要与 “传统” 对象编程不同的风格，所以在使用 actor 时要记住几点。
第一、actor 的主要能力来源于消息传递风格，而不采用阻塞-调用风格，这是它的主要特点。（有意思的是，也有使用消息传递作为核心机制的面向对象语言。最知名的两个例子是 Objective-C 和 Smalltalk，还有 ThoughtWorker 的 Ola Bini 新创建的 Ioke）。如果创建直接或间接扩展 Actor 的类，那么要确保对对象的所有调用都通过消息传递进行；
第二、因为可以在任何时候交付消息，而且更重要的是，在发送和接收之间可能有相当长的延迟，所以一定要确保消息携带正确地处理它们所需的所有状态。这种方式会： 让代码更容易理解（因为消息携带处理所需的所有状态），减少 actor 访问某些地方的共享状态的可能性，从而减少发生死锁或其他并发性问题的机会；
第三、actor 应该不会阻塞，您从前面的内容应该能够看出这一点。从本质上说，阻塞是导致死锁的原因；代码可能产生的阻塞越少，发生死锁的可能性就越低。
很有意思的是，如果您熟悉 Java Message Service (JMS) API，就会发现我给出的这些建议在很大程度上也适用于 JMS — 毕竟，actor 消息传递风格只是在实体之间传递消息，JMS 消息传递也是在实体之间传递消息。它们的差异在于，JMS 消息往往比较大，在层和进程级别上操作；而 actor 消息往往比较小，在对象和线程级别上操作。如果您掌握了 JMS，actor 也不难掌握。

注：本学习笔记来自DT大数据梦工厂        微信公众号：DT_Spark        每晚8点YY永久直播频道：68917580

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：SCALA Spark SPAR Park SPA Spark scala DT_Spark 大数据