复杂业务下,我们为何选择 Akka 作为异步通信框架？

• 容错：由于门店数量众多且情况各不相同，仍然有很多潜在的因素可能导致流程出错或失败。原则上，某次流程的失败不应该对其他流程造成任何影响，每个流程在平台层面应该成为互相独立的任务。

因此，我们需要一套轻量化的分布式服务框架，来实现满足上述需求的模型训练预测平台，并在一定程度上保证平台的可拓展性。结合此前团队内的技术积累，最终选择了 Akka 框架用于实现平台的内部通信。

• 指定接受到下一个消息时的行为

上述操作没有顺序执行的假设，因此可以并行进行。发送者与已经发送的消息解耦，可以进行无需等待的异步通信。

创建 ActorSystem 时，默认将在 classpath 中寻找 application.conf、 application.json 和 application.properties，并自动加载：

val system=ActorSystem("RsModelActorSystem")
val system=ActorSystem("RsModelActorSystem", ConfigFactory.load()) // 同上复制代码

如果想要使用自己的配置文件，可以通过 ConfigFactory 来配置加载：

val system = ActorSystem("UniversityMessageSystem", 
           ConfigFactory.load("own-application.conf")) 


        val config = ConfigFactory.parseString(
              s"""
            |akka.remote.netty.tcp.hostname = $host
            |akka.actor.provider = akka.remote.RemoteActorRefProvider
            |akka.remote.enabled-transport = akka.remote.netty.tcp
            |akka.remote.netty.tcp.port = 2445
              """.stripMargin)
        val system = ActorSystem("RsModelActorSystem",
            config.withFallback(ConfigFactory.load())) // 同上
复制代码

ActorSystem 的配置参数中有大量参数可以自定义，需要根据实际需要修改，例如在该项目中，后期单个算法任务对象大小超过了 Akka remote 默认包大小 128000 bytes，需要修改参数 akka.remote.netty.tcp.maximum-frame-size

在 Actor 类中，主要逻辑均在 receive 方法中实现，通过偏函数方法，执行并返回对应的逻辑：

//ActorLogging 提供 Actor 内部的日志输出
        class RsActor extends Actor with ActorLogging {
              override def receive: Receive = {
                case MapMessage(parameters) =>
                      println(parameters.get("code"))

                case MapKeyMessage(parameters, key) =>
                      println(parameters.get(key))

                case StringMessage(msg) =>
                      println(msg.getBytes().length)

                case o: Object =>
                      println(o.getClass)

                case _: AnyRef =>
                      println("233")
              }
        }复制代码

生成一个可以接收消息的 Actor 实例主要有两个方法：

// 生成一个基于本地类的 Actor 实例
        val rsActor = system.actorOf(Props[RsActor], "rsActor")
        // 生成一个基于远程地址的 Actor 实例
        val rmActor = 
            system.actorSelection("akka.tcp://RsModelAkkaSystem@192.168.1.9:2445/user/rsActor")

        //  使用! 向对应的 Actor 实例发送消息
        rsActor ! StringMessage("test")
        rmActor ! MapMessage(Map("code"->"233"))复制代码

Akka 中对传递的消息内容并没有太严格要求，可以是基本数据类型，也可以是支持序列化的对象：

//scala 的 case class 便于简洁地创建消息类
        case class StringMessage(msg: String) extends Serializable
        case class MapMessage(parameters: Map[String, String]) extends Serializable
        case class MapKeyMessage(parameters: Map[String, String], key: String) extends Serializable复制代码

Akka 作为一款被广泛使用的开源工具，在实际项目中体现出了很多的优势，异步的消息驱动方式也给我们提供了一套新的思路和实现方法。

作者介绍：李天烨，TalkingData 数据科学家。毕业于东北大学，任职于 TalkingData 数据科学团队，从事数据科学自动化相关工作。

本文转自：InfoQ