AKKA学习笔记

AKKA学习笔记总结

01. AKKA

1. 介绍：

Akka基于Actor模型，提供了一个用于构建可扩展的（Scalable）、弹性的（Resilient）、快速响应的（Responsive）应用程序的平台。

2. Spark中的RPC

目前大多数的分布式架构底层通讯都是经过RPC（进程间通讯）实现的，好比Hadoop项目的RPC通讯框架，可是Hadoop在设计之初就是为了运行长达数小时的批量而设计的，在某些极端的状况下，任务提交的延迟很高，全部Hadoop的RPC显得有些笨重。

Spark 的RPC是经过Akka类库实现的，Akka用Scala语言开发，基于Actor并发模型实现，Akka具备高可靠、高性能、可扩展、分布式等特色，使用Akka能够轻松实现分布式RPC功能。

3. Actor模型

1. 介绍

Actor模型：在计算机科学领域，Actor模型是一个并行计算模型，它把actor做为并行计算的基本元素来对待：为响应一个接收到的消息，一个actor可以本身作出一些决策，如建立更多的actor，或发送更多的消息，或者肯定如何去响应接收到的下一个消息。

2. Scala中的多线程

（1） 传统的并发是经过线程（thread）来实现的。在传统的并发模型中，程序被分红若干份同时执行的任务，而且全部任务都对一块共享的内存进行操做。在传统的并发模型会引发竞争问题，能够采起锁机制避免竞争问题，但同时这可能带来死锁等问题。

（2） 在Scala中，多线程的基础就是Actor，核心思想是用消息传递来进行线程间的信息共享和同步。它是基于事件模型的并发机制，Scala是运用消息（message）的发送、接收来实现多线程的。使用Scala可以更容易地实现多线程应用的开发。

Actor模型是另外一种不一样的并发模型，它很好地解决了在传统并发模型中竞争和死锁等问题。咱们能够把一个由actor模型实现的并发程序当作是一个星系同样，星系里面有不少星球，每一个星球都是一个actor，星球之间不共享任何资源，可是它们之间有通道来相互传递信息。

每一个星球（actor）都有一个信箱来接受来自其它星球的任意信息，它会按照信息接收的顺序来处理，处理完一个信息而后接着处理下一个信息。能够按照信息类型来触发不一样的行为。

同时，每一个星球（actor）能够异步地（也能够同步，但不是这里谈论的重点）向其它任意星球发送任意消息，就是说，它发送消息以后不会等待返回信息而是直接执行接下来的操做。

好比：

object MyActor1 extends Actor{
 //从新act方法
 def act(){
 for(i <- 1 to 20){
   println("actor-1 " + i)
   Thread.sleep(1000)
  }
}
}

启动线程：

//启动Actor
MyActor1.start()

关于Actor之间的消息传递不是详情见以后的一篇学习笔记。

4. AKKA实现RPC

Scala在2.11.x版本中将Akka加入其中，做为其默认的Actor，老版本的Actor已经废弃。

1. 架构图

2. 重要类和方法

ActorSystem
在Akka中，ActorSystem是一个重量级的结构，他按需分配多个线程，因此在实际应用中，ActorSystem一般是一个单例对象，咱们可使用这个ActorSystem建立不少Actor。

在Akka中，Actor负责通讯，在Actor中有一些重要的生命周期方法。
1.preStart()方法：该方法在Actor对象构造方法执行后执行，整个Actor生命周期中仅执行一次。
2.receive()方法：该方法在Actor的preStart方法执行完成后执行，用于接收消息，会被反复执行。

3. 实现

Master类

package wrd.akka

import akka.actor.Actor
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props

class Master_old extends Actor {

  println("constructor invoked")

  override def preStart(): Unit = {
    println("preStart invoked")
  }

  //用于接收消息,sender就是发送者的代理
  def receive: Actor.Receive = {
    case "connect" => {
      println("a client connected")
      sender ! "reply"
    }

    case "hello" => {
      println("hello")
    }
  }

}

object Master_old {

  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 8888
    // 准备配置
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    //ActorSystem老大，辅助建立和监控下面的Actor，他是单例的
    val actorSystem = ActorSystem("MasterSystem", config)

    val master = actorSystem.actorOf(Props(new Master_old), "Master") //Master主构造器会执行
    master ! "hello" //发送信息
    actorSystem.awaitTermination() //让进程等待着, 先别结束
    }
}

Worker类

package wrd.akka

import akka.actor.Actor
import akka.actor.ActorSelection
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props

class Worker_old(val masterHost: String, val masterPort: Int) extends Actor {

  var master: ActorSelection = _

  //创建链接
  override def preStart(): Unit = {
    //在master启动时会打印下面的那个协议, 能够先用这个作一个标志, 链接哪一个master
    //继承actor后会有一个context, 能够经过它来链接
    master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
    //须要有/user, Master要和master那边建立的名字保持一致
    master ! "connect"
  }

  def receive: Actor.Receive = {
    case "reply" => {
      println("a reply from master")
    }
  }

}

object Worker_old {

  def main(args: Array[String]): Unit = {

    val host = "127.0.0.1"
    val port = 9999

    val masterHost = "127.0.0.1"
    val masterPort = 8888
    // 准备配置
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    //ActorSystem老大，辅助建立和监控下面的Actor，他是单例的
    val actorSystem = ActorSystem("WorkerSystem", config)
    actorSystem.actorOf(Props(new Worker_old(masterHost, masterPort)), "Worker")
    actorSystem.awaitTermination()
  }
}

暂时先记录到这儿吧，再完善。 参考：《http://blog.csdn.net/fancylovejava/article/details/24724395》