Scala併發編程

併發編程

在 Java 中，多線程訪問共享數據的時候會存在【線程安全】問題
Scala 的多線程使用了新的通訊機制
經過發送消息來通訊，沒有了共享數據，從而實現併發編程
Scala 使用的是 Akka 框架，Akka 經過 Actor 模式實現高併發
Akka 是使用 Scala 語言編寫的用法高併發的編程框架
Akka 的高併發是由 Actor 與 Actor 之間的通訊
Akka 模型
消息傳遞
併發(FIFO)
容錯
Actor 是由 ActorSystem 來建立的
ActorSystem 是單例的
Actor 是多例的
一個 JVM 中，只須要有一個 ActorSystem
類型別名

[Scala] 純文本查看 複製代碼
?
object Alias {
    def main(args: Array[String]): Unit = {
        val arr1: Array[Int] = Array(1, 2, 3, 4, 5)
        type myArr = Array[Int]
        val arr2: myArr = Array(1, 2, 3, 4, 5)
    }
/size/colorfont=-apple-system, BlinkMacSystemFont, 微軟雅黑, "}

Akka 案例

建立 maven 工程，導入依賴
[XML] 純文本查看 複製代碼
?
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/m...d">
    <modelVersion>4.0.0</modelVersion>
 
    <groupId>com.study</groupId>
    <artifactId>ScalaDemo</artifactId>
    <version>1.0-SNAPSHOT</version>
 
    <!-- 定義一下常量 -->
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <akka.version>2.4.17</akka.version>
    </properties>
 
    <dependencies>
        <!-- 添加scala的依賴 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- 添加akka的actor依賴 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
        <!-- 多進程之間的Actor通訊 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
    </dependencies>
 
    <!-- 指定插件-->
    <build>
        <plugins>
            <!-- 指定編譯scala的插件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
 
            <!-- maven打包的插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>:</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>reference.conf</resource>
                                </transformer>
                                <!-- 指定maven方法 -->
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass></mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
 

HelloActor入門案例
[Scala] 純文本查看 複製代碼
?
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
 
class HelloAcrot extends Actor {
    //type Receive = scala.PartialFunction[scala.Any, scala.Unit]
    override def receive: Receive = {
        case msg: String => println(msg)
    }
}
object HelloAcrot {
    val actorSystemName = "helloActorSystem"
    val actorName = "helloActor"
 
    def main(args: Array[String]): Unit = {
        val host = "127.0.0.1"
        val port = 8888
        // 指定配置 host + port
        val prot =
            s"""
             |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
             |akka.remote.netty.tcp.hostname = ${host}
             |akka.remote.netty.tcp.port = ${port}
            """
            .stripMargin
        // 獲取一個配置實例
        val config: Config = ConfigFactory.parseString(prot)
        // 建立一個ActorSystem
        val actorSystem: ActorSystem = ActorSystem.create(actorSystemName,config)
        val actor: ActorRef = actorSystem.actorOf(Props(new HelloAcrot()),actorName)
        actor ! "Hello Dog!"
    }
}
 

Actor 互相通訊
[Scala] 純文本查看 複製代碼
?
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
 
class Ai extends Actor {
  override def receive: Receive = {
    case str: String => {
      println(s"Ai獲取到消息${str}")
      sender() ! "Ai回覆你"
    }
  }
}
 
object Ai {
  val actorSystemName = "Ai_system"
  val actorName = "Ai_actor"
 
  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 9999
    // 指定協議
    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${host}
         |akka.remote.netty.tcp.port = ${port}
        """
        .stripMargin
    val config: Config = ConfigFactory.parseString(str)
    val actorSystem: ActorSystem = ActorSystem.create(actorSystemName, config)
    actorSystem.actorOf(Props(new Ai()), actorName)
  }
}
 

[Scala] 純文本查看 複製代碼
?
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
 
class Horse extends Actor {
  override def receive: Receive = {
    case str:String => {
      println(s"獲取到Ai的回覆:${str}")
      sender() ! "Ai，你好"
    }
  }
 
  override def preStart(): Unit = {
    val path = s"akka.tcp://${Ai.actorSystemName}@127.0.0.1:9999/user/${Ai.actorName}"
    val actorProxy: ActorSelection = context.actorSelection(path)
    println(s"獲取${Horse.actorName}代理對象成功!")
    actorProxy ! "Ai你好"
  }
}
 
object Horse {
  val actorSystemName = "Horse_system"
  val actorName = "Horse_actor"
 
  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 8888
    // 指定協議
    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${host}
         |akka.remote.netty.tcp.port = ${port}
        """
        .stripMargin
    // 獲取 ActorSystem 配置
    val config: Config = ConfigFactory.parseString(str)
    // 建立一個 ActorSystem 實例
    val actorSystem: ActorSystem = ActorSystem.create(actorSystemName, config)
    actorSystem.actorOf(Props(new Horse()), actorName)
  }
}

Spark 通訊機制

HDFS: NameNode + DataNode
Spark: Master + Workder
Spark RPC 通訊的基本機制

Master 先啓動，而後 Worker 向 Master 創建鏈接
Worker 向 Master 註冊
資源（cores,memory）
workerId
用 case class 封裝相關的信息
Master 接收 Worker 的消息後，保存 workerId
集合 Map[workerId,Info類]
Master 返回註冊成功消息
case object
workerId接收到註冊成功消息後，定時發送心跳信息
定時任務 case class(workerId)
Master 定時任務，在 Master 啓動以後當即啓動
清楚超時沒發心跳的 Worder
[Scala] 純文本查看 複製代碼
?
object RpcMessage {
}
 
"worker向master註冊的樣例類"
case class Worker2Master(workerId: String)
 
"master 返回給 worker 註冊成功消息"
case object Registered
 
"worker定時向master發送的心跳信息"
case class HeartBeat(workerId: String)
 
"master 啓動定時，定時檢測超時的master"
case object CheckWorkerStataus
 

[Scala] 純文本查看 複製代碼
?

class WorkerInfo(val workerId: String) {
  var lastHeartBeatTime: Long = _
}

 

[Scala] 純文本查看 複製代碼
?
import java.util.UUID
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
class Worker(val masterHost: String, val masterPort: String) extends Actor {
  "workerId"
  val workerId: String = UUID.randomUUID().toString.replaceAll("-", "")
 
  override def receive: Receive = {
    case Registered => {
      "worker 接收到 master 返回的註冊成功消息"
      println(s"worder(${workerId})獲取到master的響應，註冊成功！")
      "worker 定時向 master 發送心跳"
      import scala.concurrent.duration._ "導入時間單位"
      import context.dispatcher
      context.system.scheduler.schedule(0 second, 10 second, sender(), HeartBeat(workerId))
    }
  }
 
  override def preStart(): Unit = {
    val path = s"akka.tcp://${Master.actorSystemName}@${masterHost}:${masterPort}/user/${Master.actorName}"
    val masterProxy: ActorSelection = context.actorSelection(path)
    println(s"worker(${workerId})建立，成功獲取 master 代理")
    masterProxy ! Worker2Master(workerId)
  }
}
 
object Worker {
  val actorSystemName = "Worker_System"
  val actorName = "Worker_Actor"
 
  def main(args: Array[String]): Unit = {
    "參數校驗"
    if (args.length != 4) {
      println("Usage:com.study.sparkrpc.Worker<master host><master port><host><port>")
      sys.exit(1)
    }
    val Array(masterHost, masterPort, host, port) = args
    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${host}
         |akka.remote.netty.tcp.port = ${port}
        """
        .stripMargin
    val config: Config = ConfigFactory.parseString(str)
    val actorSystem: ActorSystem = ActorSystem.create(actorSystemName, config)
    actorSystem.actorOf(Props(new Worker(masterHost, masterPort)), actorName)
  }
}
 

[Scala] 純文本查看 複製代碼?import akka.actor.{Actor, ActorSystem, Props}import com.typesafe.config.{Config, ConfigFactory} import scala.collection.mutable class Master extends Actor {  "存儲全部的workerInfo"  val workerMap = new mutable.HashMap[String, WorkerInfo]()   override def preStart(): Unit = {    import scala.concurrent.duration._    import context.dispatcher    context.system.scheduler.schedule(0 second, 13 second, self, CheckWorkerStataus)  }   override def receive: Receive = {    case Worker2Master(workerId) => {      println(s"master 接收到 worker(${workerId}) 的註冊")      val info = new WorkerInfo(workerId)      info.lastHeartBeatTime = System.currentTimeMillis()      workerMap.put(workerId, info)      sender() ! Registered    }    case HeartBeat(workerId) => {      workerMap(workerId).lastHeartBeatTime = System.currentTimeMillis()      println(s"master 接收到 worker(${workerId})的心跳")    }    case CheckWorkerStataus => {      println("master 檢查過時的 worker")      val deadWorks: mutable.HashMap[String, WorkerInfo] = workerMap.filter(tp => System.currentTimeMillis() - tp._2.lastHeartBeatTime > 30 * 1000)      deadWorks.foreach(tp => workerMap.remove(tp._1))      "workerMap --= deadWorks.keys"      println(s"master 的大小爲：${workerMap.size}")    }  }} object Master {  val actorSystemName = "Master_System"  val actorName = "Master_Actor"   def main(args: Array[String]): Unit = {    "參數校驗"    if (args.length != 2) {      println("Usage:com.study.sparkrpc.Master<masterHost><masterPort>")      sys.exit(1)    }    val Array(host, port) = args    val str =      s"""         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"         |akka.remote.netty.tcp.hostname = ${host}         |akka.remote.netty.tcp.port = ${port}        """        .stripMargin    val config: Config = ConfigFactory.parseString(str)    val actorSystem: ActorSystem = ActorSystem.create(actorSystemName, config)    actorSystem.actorOf(Props(new Master()), actorName)  }}