Akka Scala编程实践，轻松开发多线程、分布式以及集群式程序

Akka基础

• 参照： http://www.importnew.com/16479.html 
• Akka笔记之Actor简介

　　  Akka中的Actor遵循Actor模型。你可以把Actor当作是人。这些人不会亲自去和别人交谈。他们只通过邮件来交流。

　　  1. 消息传递 2. 并发 3. 异常处理 4. 多任务 5. 消息链

• Akka笔记之消息传递

        消息发送给actor代理；

        消息是不可变对象（可带有属性的case class）；

        分发器dispatcher和邮箱: dispatcher从actorRef取出一条消息放在目标actor邮箱中，然后放mailbox放在一个Thread上；当MailBox的run方法运行的时候，它会从队列中取出一条消息， 然后将它传给Actor去处理。在Actor的世界中，邮箱一有机会就会要求Actor去完成自己的任务。

• Akka笔记之日志及测试

        使用slf4j打印日志：Akka通过一个叫做ActorLogging的特质(trait)来实现的这一功能。可以这个trait混入（mixin）到类中。当我们要打印一条消息的时候，ActorLogging中的日志方法会将日志信息发布到一个EventStream流中。没错，我的确说的是发布。

        EventStream：EventStream就像是一个我们用来发布及接收消息的消息代理。它与常见的消息中间件的根本区别在于EventStream的订阅者只能是一个Actor。DefaultLogger默认订阅这些消息并打印到标准输出。

akka{ 
    loggers = ["akka.event.slf4j.Slf4jLogger"]
    loglevel = "DEBUG"
    logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
}

• Akka笔记之请求与响应

        技术上来讲，消息发送给Actor就是希望能有副作用的。设计上便是如此。目标Actor可以不做响应，也可以做如下两件事情——

        1. 给发送方回复一条响应（在本例中，TeacherActor会将一句名言回复给StudentActor）

        2. 将响应转发给其它的目标受众Actor，后者也可以进行响应/转发/产生副作用。Router和Supervisor就是这种情况。

• Akka笔记之配置及调度

　　　配置管理：applicaiton.conf

        调度器: 一次调度和循环调度。import context.dispatcher这条语句非常重要。schedule方法需要一个非常重要的隐式参数——ExecutionContext。schedule方法只是把消息发送封装到了一个Runnable中，而它最终是由传进来的ExecutionContext来执行的。

• Akka笔记之生命周期

        preStart: Actor重启的时候（比如说崩溃了之后重启）通过调用preStart方法还能重新初始化。而构造方法则实现不了这点（只会初始化一次）。

        postStop: ActorSystem.stop(), ActorContext.stop(), PoisonPill 都可以终止一个actor，关闭时回调用postStop()

• Akka笔记之子Actor及路径

        Actor是纯粹的分层结构。你所创建出来的Actor必定是某个Actor的子Actor。actorRef.path可以获取到actor路径。

        子Actor：当某个任务由一个或多个子任务所组成的时候通常就会创建子Actor。或者当某个任务由父Actor执行比较容易出错，而你希望将它进行隔离的时候，也可以使用子Actor（这样当子Actor崩溃的时候，你还能够恢复它）。如果不存在父子Actor关系，就不要创建子Actor。

• Akka笔记之Actor监控

        监控(Watch)：不管Actor是怎么挂掉的，系统里面会有些Actor希望能够知晓这一情况。ActorContext.watch和ActorContext.unwatch就是监控与取消监控的方法了。进行了监控之后，监控者会收到已停止的Actor发来的一条Terminated消息，它们只需要把这个消息放到receive函数的处理逻辑里就好了。 

        监督（Supervision）：只存在于父子关系的actor之间。

Actor模型

Actor模型的要点

• 基于Actor的系统中 Actor是最小的抽象单元，  就像object 之于 oop
• 一个Actor封装了状态和行为
• 外界不能进入Actor以获取其状态、字段、执行方法，和Actor的交互只能通过message
• 一个Actor有一个信箱mailbox，将外部发送来的消息msg存到队列中
• Actor的终生就是在等待msg，并依次取出mailbox中的消息进行处理

Actor模型的组织观

• 将Actor系统视为一个公司，从上到下有严格的层级关系，公司中员工是person，也是actor
• 一个Actor仅有一个上级主管 称为 supervisor，就是create这个Actor的 那个Actor
• 一个Actor可能有多个下属小兵，真正干着脏累差活的actor
• 一个Actor可能有多个同级的兄弟部门actor

开发actor系统的关键

• 委托 委托 委托！！！ 以做的更多！
• 公司老总不能事无巨细全部承担，他的主要工作就是分配公司工作和监督
• 各部门主管收到任务，要么自己干，要么再细分任务，然后分配到自己的下属手中并监督
• 如果公司还有更多的层级，则继续上面这个主管的龌龊勾当
• 最底层的员工收到细粒度任务，干着自己最擅长的事情

Actor的失效处理

• 人无完人，Actor也不可能100%完成任务
• 如果任务执行失败，一个Actor会挂起自己及其下属，然后发消息告诉其上级主管“我失败了”！
• 上级主管收到下属的“失败”消息时，可以有如下反应：

1. 失败就失败，就这样吧，没太大关系：保持当前状态，恢复Actor，继续工作
2. 没成功啊，那重新做一遍吧             ：重置状态，重启该Actor
3. 没机会了，失败了你就滚！解雇下属：关闭 终结 该 Actor
4. 这事我也决定不了，我请示我的上级：向上级Actor报告

Akka实现的Actor模型的一些附加特性

• 实例化一个Actor时，返回一个ActorRef，相当于 邮箱地址，并不能通过这个获取Actor的状态信息
• Actor模型是线程的更高层抽象，最终是跑在java的线程中的
• 多个Actor可能共享一个线程，这是由akka保障的
• Actor的信箱有多种实现方式：无限变量信箱    有限信箱    带优先级信箱， 还可以自定义实现
• akka没有让Actor扫描信箱的message
• 一个Actor终结（无论是正常还是非常的）它的信箱中的msg进入akka系统的“死信箱dead letter mailbox”中

Akka framework现在已经是Scala语言的一部分了，用它编写分布式程序是相当简单的，本文将一步一步地讲解如何做到scale up & scale out。

• 简单的单线程程序

    先从一个简单的单线程程序PerfectNumber.scala开始，这个程序是找出2到100范围内所有的“完美数”（真约数之和恰好等于此数自身）

package com.newegg.demo

import scala.concurrent.duration._
import scala.collection.mutable.ListBuffer

object PerfectNumber {
  def sumOfFactors(number: Int) = {
    (1 /: (2 until number)) { (sum, i) => if (number % i == 0) sum + i else sum }
  }

  def isPerfect(num: Int): Boolean = {
    num == sumOfFactors(num)
  }

  def findPerfectNumbers(start: Int, end: Int) = {
    require(start > 1 && end >= start)
    val perfectNumbers = new ListBuffer[Int]
    (start to end).foreach(num => if (isPerfect(num)) perfectNumbers += num)
    perfectNumbers.toList
  }

  def main(args: Array[String]): Unit = {
    val list = findPerfectNumbers(2, 100)
    println("
Found Perfect Numbers:" + list.mkString(","))
  }
}

• 多线程程序

    Scala编写并发程序的基础是Actor模型，与Actor交互的唯一途径是“消息传递”，你根本不用考虑“进程”，“线程”，“同步”，“锁”等等一些冷冰冰的概念，你可以把Actor看做是一个“人”，你的程序是一个“组织”内的一群“人”之间以“消息传递”的方式在协作。
    这个示例中要用到的“消息”定义在Data.scala文件中，内容如下：

package com.newegg.demo

import akka.actor.ActorRef

sealed trait Message
case class StartFind(start: Int, end: Int, replyTo: ActorRef) extends Message
case class Work(num: Int, replyTo: ActorRef) extends Message
case class Result(num: Int, isPerfect: Boolean) extends Message
case class PerfectNumbers(list: List[Int]) extends Message

    用面向对象的方式把程序改造一下，把PerfectNumber.scala其中的部分代码抽取到一个单独的Worker.scala文件中：

package com.newegg.demo

import akka.actor.Actor
import akka.actor.ActorRef
class Worker extends Actor {
  private def sumOfFactors(number: Int) = {
    (1 /: (2 until number)) { (sum, i) => if (number % i == 0) sum + i else sum }
  }

  private def isPerfect(num: Int): Boolean = {
    num == sumOfFactors(num)
  }

  def receive = {
    case Work(num: Int, replyTo: ActorRef) =>
      replyTo ! Result(num, isPerfect(num))
      print("[" + num + "] ")
  }
}

一部分代码抽取到Master.scala文件中：

package com.newegg.demo

import scala.collection.mutable.ListBuffer
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Props
import akka.routing.FromConfig
import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope

sealed class Helper(count: Int, replyTo: ActorRef) extends Actor {
  val perfectNumbers = new ListBuffer[Int]
  var nrOfResult = 0

  def receive = {
    case Result(num: Int, isPerfect: Boolean) =>
      nrOfResult += 1
      if (isPerfect)
        perfectNumbers += num
      if (nrOfResult == count)
        replyTo ! PerfectNumbers(perfectNumbers.toList)
  }
}

class Master extends Actor {
  val worker = context.actorOf(Props[Worker].withRouter(FromConfig()), "workerRouter")

  def receive = {
    case StartFind(start: Int, end: Int, replyTo: ActorRef) if (start > 1 && end >= start) =>
      val count = end - start + 1
      val helper = context.actorOf(Props(new Helper(count, replyTo)))
      (start to end).foreach(num => worker ! Work(num, helper))
  }
}

    这里用到了一个“可变”的变量nrOfResult，有时候，要完全不用“可变”的变量是相当难以做到的，只要将“可变”的副作用很好地进行“隔离”还是可以的。Scala语言既提倡使用“不变”变量，也容忍使用“可变”变量，既提倡“函数式”编程风格，也兼容面向对象编程，它并不强迫你一开始就完全放弃你所熟悉的编程习惯，我很喜欢这种比较中庸的语言。

    那个单线程程序的主函数改造如下：

package com.newegg.demo

import scala.concurrent.duration._
import scala.collection.mutable.ListBuffer
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.Actor
import com.typesafe.config.ConfigFactory
import akka.routing.FromConfig

object PerfectNumber {

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("MasterApp", ConfigFactory.load.getConfig("multiThread"))
    system.actorOf(Props(new Actor() {
      val master = context.system.actorOf(Props[Master], "master")
      master ! StartFind(2, 100, self)
      def receive = {
        case PerfectNumbers(list: List[Int]) =>
          println("
Found Perfect Numbers:" + list.mkString(","))
          system.shutdown()
      }
    }))
  }
}

    程序中用到的配置application.conf文件中的内容如下：

multiThread{
  akka.actor.deployment./master/workerRouter{
    router="round-robin"
    nr-of-instances=10
  }
}

    这样，单线程程序就完全改造成了一个可以充分利用计算机上所有的CPU核的多线程程序，根据计算机的硬件能力只需调整nr-of-instances配置参数就可以调整并发的能力。

• 分布式程序

   现在，我们进一步改造，把它变成一个可以跨JVM，或者说跨计算机运行的分布式程序。

新建一个MasterApp.scala文件：

package com.newegg.demo

import com.typesafe.config.ConfigFactory
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSelection.toScala
import akka.actor.ActorSystem
import akka.actor.Props
import akka.kernel.Bootable
import akka.cluster.Cluster

class Agent extends Actor {
  var master = context.system.actorSelection("/user/master")

  def receive = {
    case StartFind(start: Int, end: Int, replyTo: ActorRef) if (start > 1 && end >= start) =>
      master ! StartFind(start, end, sender)
  }
}

class MasterDaemon extends Bootable {
  val system = ActorSystem("MasterApp", ConfigFactory.load.getConfig("remote"))
  val master = system.actorOf(Props[Master], "master")

  def startup = {}
  def shutdown = {
    system.shutdown()
  }
}

object MasterApp {
  def main(args: Array[String]) {
    new MasterDaemon()
  }
}

application.conf文件中加入一个remote的section配置块：

akka {
  actor {
    provider = "akka.remote.RemoteActorRefProvider"
    deployment{
      /remoteMaster{
        router="round-robin"
        nr-of-instances=10
        target{
          nodes=[
            "akka.tcp://MasterApp@127.0.0.1:2551",
            "akka.tcp://MasterApp@127.0.0.1:2552"
          ]
        }
      }
      /master/workerRouter{
        router="round-robin"
        nr-of-instances=10
      }
    }
  }

  remote {
    transport = "akka.remote.netty.NettyRemoteTransport"
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2551
    }
  }
}

    在Terminal中运行命令java -cp ".:../../lib/*" com.newegg.demo.MasterApp，可以看一个守护程序正在监听2551端口，修改上述配置端口为2552，在另一个Terminal中运行同样的命令，另一个守护程序正在监听2552端口。
    修改PerfectNumber.scala中的main函数为：

def main(args: Array[String]): Unit = {
   val system = ActorSystem("MasterApp", ConfigFactory.load.getConfig("remote"))
   system.actorOf(Props(new Actor() {
     val agent = context.system.actorOf(Props(new Agent()).withRouter(FromConfig()), "remoteMaster")
     dispatch

 private def dispatch = {
   val remotePaths = context.system.settings.config.getList("akka.actor.deployment./remoteMaster.target.nodes")
   val count = end - start + 1
   val piece = Math.round(count.toDouble / remotePaths.size()).toInt
   println("%s pieces per node".format(piece))
   var s = start
   while (end >= s) {
     var e = s + piece - 1
     if (e > end)
       e = end
     agent ! StartFind(s, e, self)
     s = e + 1
   }
   println(agent.path)
 }

     def receive = {
       case PerfectNumbers(list: List[Int]) =>
         println("
Found Perfect Numbers:" + list.mkString(","))
         system.shutdown()
     }
   }))
 }

修改配置端口为2553，在Terminal中运行命令java -cp ".:../../lib/*" com.newegg.demo.PerfectNumber，可以看到两个守护程序中均有Worker在工作，总的计算任务得到了分担。
    这种分布式程序实现起来简单，但是有个缺点：参与分担任务的守护程序的地址必须全部在target.nodes配置中列出，且一旦其中有守护程序宕掉，整体将不能正确地对外服务。
    因此，我们需要有一个有更具扩展能力和高容错能力的集群式应用。

• 集群式应用
  

    在application.conf文件中新加一个cluser配置区块：

akka {
  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
    deployment {
      /master/workerRouter {
        router = "consistent-hashing"
        nr-of-instances = 10
            cluster {
              enabled = on
              max-nr-of-instances-per-node = 3
              allow-local-routees = on
            }
          }
        }
  }
  remote {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2551
    }
  }
  cluster {
    min-nr-of-members = 2
    seed-nodes = [
      "akka.tcp://MasterApp@127.0.0.1:2551",
      "akka.tcp://MasterApp@127.0.0.1:2552"]

    auto-down=on
  }
}

注意其中有两个seed-nodes，是集群的“首脑”，某节点会加入的是哪个集群，正是因为参照这个seed-nodes来的。
    这里采用的是consistent-hashing Router，集群节点之间有心跳检测，集群实现中内部采用的是与Cassandra一样的Gossip协议，用一致性哈希来维护集群节点“环”。
    改造Master.scala文件中其中一行代码，将

worker ! Work(num, helper)

改为:

worker.tell(ConsistentHashableEnvelope(Work(num,helper), num), helper)

这样，发送到集群中的消息支持一致性哈希，在整个集群节点中分散任务。 
    改造上面的MasterApp.scala，在其中代码行val master = system.actorOf(Props[Master], "master")下面增加两行代码：

val agent = system.actorOf(Props(new Agent), "agent")
 Cluster(system).registerOnMemberUp(agent)

将分布式的守护程序改成了集群式的守护程序（其实这段代码没必要放到Bootable类中），以上述同样的方式运行MasterApp，以不同的端口，跑起任意个守护程序，它们均会join到同一集群中，只要有一个seed-nodes存在，集群就能正常对外提供服务。
    新建一个ClusterClient.scala文件，内容如下：

package com.newegg.demo

import scala.concurrent.forkjoin.ThreadLocalRandom

import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSelection.toScala
import akka.actor.Address
import akka.actor.RelativeActorPath
import akka.actor.RootActorPath
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.ClusterEvent.MemberEvent
import akka.cluster.ClusterEvent.MemberRemoved
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.MemberStatus

class ClusterClient extends Actor {
  val cluster = Cluster(context.system)
  override def preStart(): Unit = cluster.subscribe(self, classOf[MemberEvent])
  override def postStop(): Unit = cluster unsubscribe self

  var nodes = Set.empty[Address]

  val servicePath = "/user/agent"
  val servicePathElements = servicePath match {
    case RelativeActorPath(elements) => elements
    case _ => throw new IllegalArgumentException(
      "servicePath [%s] is not a valid relative actor path" format servicePath)
  }

  def receive = {
    case state: CurrentClusterState =>
      nodes = state.members.collect {
        case m if m.status == MemberStatus.Up => m.address
      }
    case MemberUp(member) =>
      nodes += member.address
    case MemberRemoved(member, _) =>
      nodes -= member.address
    case _: MemberEvent => // ignore
    case PerfectNumbers(list: List[Int]) =>
      println("
Found Perfect Numbers:" + list.mkString(","))
      cluster.down(self.path.address)
      context.system.shutdown()
    case StartFind(start: Int, end: Int, resultTo: ActorRef) =>
      println("node size:" + nodes.size)
      nodes.size match {
        case x: Int if x < 1 =>
          Thread.sleep(1000)
          self ! StartFind(start, end, resultTo)
        case _ =>
          val address = nodes.toIndexedSeq(ThreadLocalRandom.current.nextInt(nodes.size))
          val service = context.actorSelection(RootActorPath(address) / servicePathElements)
          service ! StartFind(start, end, resultTo)
          println("send to :" + address)
      }
  }
}

上面的代码是比较固定的写法，此Actor的作用是：加入集群，订阅集群中有关节点增减变化的消息，维护一个集群中存活节点的地址列表，将任务消息发到集群节点中。

    改造上述main函数如下：

val system = ActorSystem("MasterApp", ConfigFactory.load.getConfig("cluster"))
   system.actorOf(Props(new Actor() {
     context.system.actorOf(Props[ClusterClient], "remoteMaster") ! StartFind(2, 100, self)
...

运行此集群客户端程序，可以看到客户端也join到集群中，集群中所有的节点都在分担计算任务，任意增减集群节点数目都是如此。