注:本文大部分文字摘抄《Akka 实战:快速构建高可用分布式应用》中内容,但此书中所用的akka版本比较老,所以本文中的代码是本文作者基于最新版本的akka以及官网所实现。
文中涉及到的所有Demo都在github上
Akka简介
Akka 是一款高性能、高容错性的分布式&并行应用框架,遵循 Apache 2 开源许可,底层通过 JVM 上另外一个流行的语言 Scala 实现,提供Java&Scala API 。它基于经典的 Actor 并发模型(即所有的消息都是基于 Actor 组件进行传递),拥有如下特点:
并行与并发:提供对并行与并发的高度抽象。
异步非阻塞:Akka-Actor 消息通信都是基于异步非阻塞。
高容错性:为跨多 JVM 的分布式模型提供强劲的容错处理,号称永不宕机。
持久化:Actor 携带的状态或消息可以被持久化,以便于在 JVM 崩溃后能恢复状态。
轻量级:每个 Actor 大约只占 300bytes,即 1G 内存可容纳接近 300 万个 Actor。
基本上,Akka 从底层就解决了我们大多数分布式&并行程序常见的难题,让工程师更专注于业务实现,同时,它也保留了多个扩展接口及配置,便于满足个性化定制的需要!
Akka应用场景
目前 Akka 已经在多家互联网&软件公司广泛使用,比如 eBay、Amazon、VMWare、PayPal、阿里、惠普、豌豆荚等,涉及行业包括游戏、金融投资、医疗保健、数据分析等。
使用场景包括:
服务后端:比如 rest web,websocket 服务,分布式消息处理等。
并发&并行:比如日志异步处理,密集数据计算等。
总之,对高并发和密集计算的系统,Akka 都是适用的!
Akka架构体系
Akka 采用 Scala 开发,运行于 JVM 之上,提供了 Scala 和 Java 两种 API,目前属于 Lightbend 公司(原名 Typesafe)。它实现了经典的 Actor 模型,同时也提供了丰富的组件,比如邮箱(MailBox)、路由(Routing)、持久化( Persistence )、网络(包括远程、集群)等,在底层对分布式&并行模式进行了高度且统一的抽象,使工程师用很少的代码就可以实现一个完整的分布式应用。
Actor模型
Actor 模型最早在 1973 年由Carl Hewitt 提出,它高度抽象了分布式并行程序的运行模式,从底层屏蔽了线程和锁机制的管理,为开发者提供了简单可依赖的开发方式。
Actor 模型认为,并行计算的最小单元就是一个 Actor 实例,而每个实例拥有自己的状态和行为,在一个大型系统中,可能存在成千上万个 Actor 实例,它们之间通过消息的方式进行通信,每个 Actor 都能发送消息给其他 Actor,也能从其他 Actor 接收消息。当我们在执行某个计算任务时,会给对应的 Actor 实例发送一个相关的消息,该 Actor 在接收消息后开始执行计算任务,由于整个消息通信的过程是异步的,所以不用等到 Actor 执行完整个过程就能执行下一步(发送消息后会马上返回),这种异步通信的方式大大提高了程序的响应性。
体系结构
Actor 是 Akka 最核心的概念,也是最基本的执行单元,所以对 Actor 管理和监控的有效性是极为重要的。在 Akka 中,每个 Actor 都有自己的监管对象,即该 Actor 的创建者,它们通常会负责子 Actor 的失败处理,另外,某些 Actor 也需要对生命周期进行监控(比如该 Actor 的终止),以便及时响应并做正确处理,这些监督和监控者本身也都是一个 Actor。在 Akka 中,整个 Actor 体系被抽象成一个 ActorSystem ,它是一个层级的结构,拥有公共行为的配置和管理。
在ActorSystem 基础上,Akka 也提供了一些配套的组件,比如持久化,HTTP 服务,网络服务等,它们都是构建高可用分布式应用不可或缺的部分,基本架构体系和周边产品如下图所示。
Actor组件
在 Akka 中,Actor 是一个高度抽象的对象引用,它包含以下几个要素:
引用( Actor Reference ):Actor 的引用不同于普通对象的引用,也不能通过传统「new」的方式直接创建一个 Actor 对象。很多时候需要通过 actorOf 或者 actor-Selection 等方式返回一个ActorRef 对象,该对象有可能存在于本地,也可能存在于远程节点,对我们来说,它是位置透明的。Actor 之间的通信和执行任务都是通过消息驱动的(而不是 API 的调用)。
状态(State):Actor 在不同时刻可能有着不同的状态,这些状态用变量来表示。在底层实现上,Actor 是运行于线程池之上的,肯定会存在多个 Actor 共享同一个线程的情况,那么会不会出现并发问题呢?实际上,Akka 为每个 Actor 都抽象出一个轻量级的执行「线程」(不是真的线程),在底层已经实现了隔离,所以基本上不用担心该问题的出现。另外,当 JVM 崩溃时,为了避免 Actor 状态的丢失,我们可以借助持久化方案来对状态进行持久化操作。
行为(Behavior):Actor 有接收和发送消息的能力,每当它接收一条消息后,就可以执行某个业务操作,同时也可以把消息转发到其他节点进行处理。
监管策略( Supervisor Strategy ):Actor 系统是一个层级结构,当任务被某个 Actor 分摊到子 Actor 时,父 Actor 就拥有监管子 Actor 的义务。在监管时,我们需要根据不同的情况选择不同的处理方案(比如停止、重启、恢复或者失败上溯)和策略(比如 1 vs 1、1 vs N 策略)。
邮箱(Mailbox)
每个 Actor 都有自己的邮箱,所有其他 Actor 发送过来的消息都会进入该邮箱。Akka 自带多种邮箱类型,也提供自定义邮箱的接口。
路由(Routing)
消息除了通过普通的 Actor 发送之外,也可以通过路由发送。当通过路由发送消息时,我们可以根据需求来选择不同的路由策略,比如轮询、广播等。路由也可以是一个 Actor。
状态持久化(Persistence )
任何程序都有失败的可能,即便是 JVM 如此强大稳定的平台也一样。当程序出错、JVM 崩溃时,任何关键状态的丢失,对我们后续的业务来讲都可能是致命的打击,所以状态数据的持久化变得非常重要。Akka 提供了 Actor 状态的持久化方案,以便我们在必要时恢复数据。
网络(远程和分布式集群)
网络功能是实现远程 Actor 和分布式集群的基础,这其中包含 I/O、网络通信(TCP/UDP)、序列化配置、分布式通信协议(Gossip)、节点(node)管理、集群分片等内容。
HTTP 模块
Akka 提供了简单易用的 HTTP 模块,支持完整的 HTTP 服务端与客户端开发,可以帮助我们快速构建性能极强的Rest Web 服务。
相关开源项目
Akka 具有高性能、可扩展、设计友好等诸多优点,非常适合用来作为分布式应用的基础框架,而且由于对 HTTP 有非常好的支持,也让它在 Web 服务领域占有一席之地。目前业界已经有多个基于 Akka 实现的开源项目,项目类型涵盖了 Web 开发、微服务、分布式文件或计算服务等。下面是 Akka 中两个具有代表性的开源项目:
Play 框架:一款大名鼎鼎的 Web 开发框架。它不同于其他 Servlet 系的框架,比如 Struts 或者SpringMVC ,底层基于 Akka 构建,天生拥有异步的特点,具有极佳的性能。它默认提供 RESTful 风格的 API,同时也对WebSocket 有不错的支持。
Lagom 框架:在目前 IT 界,最火爆的概念要属「微服务」了,微服务的理念是,把业务功能拆成小的、独立的单元,它们之间能够互相通信而且支持水平扩展。Lagom 就是这样一款微服务框架,它基于异步的消息驱动,对分布式集群、持久化(如 JPA、NoSQL)都有良好的支持。同时,它也拥有完整的集成开发环境,非常便于在线部署和管理。
基本使用
环境搭建
JDK版本最好是在 1.6 以上,不过后续会介绍Akka的一些新特性,所以最好采用1.8的版本。
Maven依赖
使用最新版akka依赖。
<dependency><groupId>com.typesafe.akka</groupId><artifactId>akka-actor_2.12</artifactId><version>2.5.23</version></dependency>
注意Akka中不同组件分为了不同的模块,我们这里暂时只是引用了核心模块,通过它可以使用akka的最基本功能。除开核心模块还包括akka-persistence 、akka-remote 、akka-cluster 等模块,分别用于实现持久化、远程、集群等功能。
创建一个Actor
publicclassActorDemoextendsAbstractActor{privateLoggingAdapterlog=Logging.getLogger(this.getContext().getSystem(),this);@OverridepublicReceivecreateReceive() {returnreceiveBuilder() .match(String.class,s->log.info(s)) .matchAny(o->{log.info("any"+o.toString()); }) .build(); }publicstaticvoidmain(String[]args) {ActorSystemsystem=ActorSystem.create("sys");ActorRefactorRef=system.actorOf(Props.create(ActorDemo.class),"actorDemo");actorRef.tell("hello world",ActorRef.noSender());// ActorRef.noSender()实际上就是叫做deadLetters的actor}}
通常每个应用程序只需要创建一个ActorSystem对象
在创建 ActorSystem 和 Actor 时,建议都指定名字,如果不知道akka将自动生成,在该示例中,它们的名字分别是 sys 和 actorDemo ,在同一个 ActorSystem 中 Actor 不能重名。
actorOf方法返回的不是Actor对象本身,而是ActorRef,表示Actor对象的引用。我们在向某个Actor发送消息时都是通过该引用进行的。
在Actor内部,可以使用getContext()来创建该Actor的子Actor,比如ActorRef childActor=getContext().actorOf(Props.create(ChildActor.class),"childActor" );
工厂模式
ActorSystem 和 ActorContext 通过接收一个 Props 实例来创建 Actor,而 Props 实例本身有两种方式可以创建:
Props.create(ActorDemo.class, Object... args)
指定一个Props工厂。
对于比较简单的场景使用第一种就足够了,而如果需要按统一的规则创建Actor,我们可以在Actor内定义一个创建Props的方法,比如:
publicstaticPropscreateProps() {returnProps.create(newCreator<Actor>() {@OverridepublicActorcreate()throwsException{returnnewHelloWorldActor(); } }); }
然后可以使用一下代码来创建Actor:
ActorRefactorRef=system.actorOf(ActorDemo.createProps(),"actorDemo");
可以看出,如果我们使用工厂模式,在创建Actor时调用工厂方法,我们不用关心Actor创建的具体过程了,有Actor自己封装即可。
发送与接收消息
Actor中的createReceive方法是用来接收并处理消息的。
createReceive方法需要放回Receive对象,而Receive对象就是真正用来处理消息的逻辑。
Receive对象可以定义规则使接收到的消息进入不同的处理逻辑:
match(Class
type):消息类型为指定type的进入此分支
matchEquals(P object):消息值与object相等的进入此分支
matchAny():任何消息都可以进入到此分支
当然,一个消息只会进入到一个分支,如果一个消息匹配多个条件优先进入定义靠前的条件。
可以使用tell和ask两种方式发送消息,它们都以异步的方式发送消息,不同的是,前者发完后立即返回,而后者期待得到一个返回结果,假如在设置的时间(Timeout)内没有得到返回结果,发送方会收到一个超时异常。
tell方法
第一个参数即"消息",它可以是任何可序列化的数据或对象,第二个参数表示发送者,noSender()表示无发送者(实际上是一个叫做 deadLetters 的 Actor,后面介绍)。假如想在Actor内部得到发送者,可以调用getSender ()方法。
在调用 tell 方法后,Actor 就会异步的去处理该消息,并不会阻塞后续代码的运行。
ask方法
这是一种“请求-响应”模式,ask方法会将返回结果包装在scala.concurrent.Future中。
对于一个Actor如何将响应返回给消息发送者呢,其实就是想响应作为消息发送给请求消息的发送者:
getSender().tell("hello "+msg,getSelf());
使用样例:
publicclassAskDemoextendsAbstractActor{@OverridepublicReceivecreateReceive() {returnreceiveBuilder().matchAny(o->{System.out.println("发送者是"+getSender());Thread.sleep(1000);getSender().tell("hello "+o,getSelf()); }).build(); }publicstaticvoidmain(String[]args) {ActorSystemsystem=ActorSystem.create("sys");ActorRefaskActor=system.actorOf(Props.create(AskDemo.class),"askActorDemo");Timeouttimeout=newTimeout(Duration.create(2,TimeUnit.SECONDS));Future<Object>f=Patterns.ask(askActor,"Akka Ask",timeout);System.out.println("ask ...");f.onSuccess(newOnSuccess<Object>() {@OverridepublicvoidonSuccess(Objectresult)throwsThrowable{System.out.println("收到消息:"+result); } },system.getDispatcher()); }}
消息转发
Actor可以将接收到的消息转发给其他Actor,语法为:
staticclassForwardActorextendsAbstractActor{privateActorReftarget=getContext().actorOf(Props.create(TargetActor.class),"targetActor");@OverridepublicReceivecreateReceive() {returnreceiveBuilder().matchAny((message)->{target.forward(message,getContext()); }).build(); } }
TargetActor接收到转发过来的消息后,发送者任然是消息原始的发送者,而不是转发者。
查找Actor
对于已存在的 Actor,我们可以根据路径来进行查找,API 如下:
ActorSelection as = [ActorSystem/ActorContext].actorSelection([path]);
在调用它时,必须指定一个绝对或者相对路径(包括远程路径),比如:
/user/parentActor/childActor
../parentActor/childActor
akka.tcp://testuser@127.0.0.1:2554/user/parentActor/childActor
查找结果为ActorSelection,表示一个或多个Actor,所以向ActorSelection发送消息时,对应的一个或多个Actor都将接收到消息。
消息不可变
Akka中都是消息驱动的,所以消息是非常重要。理论上,Actor 可以传递或接收任何类型的消息,但是为了避免出现竞态条件(对共享资源的并发写)和不可预知的数据篡改,最好把消息设计成不可变的对象。在该对象中,我们通常会把所有变量设计成只读,在该对象的整个生命周期内,这些数据将不可修改。所以在 Java 中,我们借助final关键字来实现。
对于集合框架,有必要通过 Collections.unmodifiable 系列的 API 来保证不变性。Collections.unmodifiable 内部使用 UnmodifiableList 、UnmodifiableMap 等 API 对现有的集合进行包装,它们通过重写其中的 set、add、put 等方法来达到限制数据修改的目的。
Actor行为切换
在业务处理过程中,往往会面临一些关键数据的变化,而这些变化也会进一步影响着程序的执行逻辑,很多时候,我们把这些关键数据称之为「状态」。状态的处理需要依赖于具体的业务,而业务本身也可能充满不确定性,一种比较好的做法是,我们把多个状态的处理过程(即行为)封装起来,形成一个个独立的组件,在使用时就可以很方便地组装、重用或切换。在 Actor 中,行为可以通过Receive来定义,一旦抽取完我们所需要的Receive,就可以使用 become/unbecome 方法来切换他们,其中 become 表示切换为某个行为,unbecome 表示修改回上一个行为。
Actor生命周期
Actor 在运行时中会经历不同的阶段,比如创建、运行、重启和销毁等,这一系列的过程或状态,我们称之为生命周期。在理想情况下,Actor 应该像一头老黄牛一样任劳任怨地不停工作,但实际情况往往是,当某个不算常态的错误发生时(比如网络偶尔超时),我们想让它重启一下,然后重复之前的动作;又或者,当一个程序异常被抛出时,我们不得不停止该 Actor,让它不能继续走下去,这些行为有可能是手动触发的,也可能是 Actor 的监督-容错机制导致的。无论怎样,当这些行为发生时,我们都希望能及时感知并做有效处理。
Actor 生命周期主要包括启动(Start)、恢复(Resume)、重启(Restart)、停止(Stop)这四个阶段,每个阶段都会伴随着自身状态信息的变化。
创建并启动
当通过 actorOf 创建并启动 Actor 时,该 Actor 除了默认拥有已指定的 path 外,也会被分配一个 UID(可以通过 getSelf().path().uid()得到该值),作为它的唯一标识。在 Actor 启动后,会默认调用preStart 方法,在该方法里,我们可以做些资源初始化的操作。
恢复运行
当 Actor 出现某种异常后,通过容错机制,可以让该 Actor 恢复并继续运行,此时 Actor 会延用之前的实例,状态也会被保留下来。
重启
当 Actor 出现某种异常后,通过容错机制,可以让该 Actor 执行重启。重启的具体过程如下:
调用旧实例的preRestart 方法,该方法默认情况下会停掉所有子级 Actor,并调用postStop 方法;
创建新实例,并在新实例上调用postRestart 方法,该方法默认情况下会调用preStart 方法。
Actor 重启后,path 和 UID 不变,这意味着假如要继续使用该 Actor,不需要重新获取ActorRef 对象,使用之前那个就可以了。同时也得注意到:Actor 重启后不会保留自身状态。
停止
当 Actor 停止时,会调用postStop 方法,同时会发送一条Terminated信息给自己的监控者,告知自己已处于停止状态。
停掉一个Actor
停止 Actor 大致有三种方式,它们分别是:
调用ActorSystem 或者getContext() 的 stop 方法;
给 Actor 发送一个PoisonPill (毒丸)消息;
给 Actor 发送一个 Kill 的消息,此时会抛出ActorKilledException 异常,并上报到父级supervisor 处理。
Actor 在停止时都会遵循一套比较可靠的流程:
当停止 Actor 时,正在处理的消息会在完全停止之前处理完毕,后续信息将不再进行处理,邮箱(用来保存 Actor 的消息)将被挂起;
给所有子级 Actor 发送终止指令,当子级都停掉后,再停掉自己,停止完毕后会调用postStop 方法,在这里可以清理或释放资源;
向生命周期监控者(DeathWatch )发送Terminated 消息,以便监控者做相应的处理。
可以使用getContext().watch();方法监控某个节点的停止信息,当该节点停止后,监控者会接收到一个Terminated消息。
监督与容错处理
Actor 系统采用"父监督"的模式进行管理,即父 Actor 会监督子 Actor 的异常情况,然后根据默认或者预设的处理逻辑来确定到底是该恢复 Actor、停止 Actor、重启 Actor 还是把错误上溯到父级。
Akka 提供了两种监督策略,分别是 One-For-One Strategy 和 All-For-One Strategy ,前者表示当一个子 Actor 出现异常时,只对该 Actor 做处理,后者则表示对所有子 Actor 都做处理,大部分时候应该选择 One-For-One Strategy (这也是默认的策略),除非子 Actor 之间的业务有很强的关联或者互相依赖。当程序中没有显式指定策略时,会启动一个默认策略:
当抛出ActorInitializationException 和ActorKilledException时,会终止子 Actor;
当抛出Exception 时,会重启子 Actor;
抛出其他类型的Throwable 异常时,会上溯失败到父级。
在 Actor 执行任务时可能会抛出不同类型的异常,很多时候,我们应该知道,哪些异常需要让 Actor 停止或重启,哪些异常可以「视而不见」。为了更加细化这种判断,我们需要自定义监督逻辑。
自定义一个监督逻辑非常简单,只需要在父 Actor 中创建一个SupervisorStrategy 对象,并通过supervisorStrategy ()方法返回出来.
熔断(circuit breaker)
当一个Actor出现异常时,如果用户还继续向这个Actor发送消息,很有可能这个消息还是无法被处理,还额外的增加了服务器的压力,Akka中提供了熔断机制(circuit breaker)。
Circuit Breaker 状态图(图片来自 Akka 官网)
circuit breaker有三个状态,分别是:Closed、Open、Half-Open。
Close 状态
正常情况下,circuit breaker是closed状态:Actor出现异常或调用超过配置的callTimeout,就增加一次失败计数;
成功则重置失败计数为 0;
当失败次数达到maxFailures后,circuit breaker会进入Open状态。
Open 状态
Actor接收到的所有调用将抛出CircuitBreakerOpenException ,立即失败;
在resetTimeout后,circuit breaker 进入Half-Open状态。
Half-Open 状态
第一次调用将尝试运行而不会快速失败;
如果第一个调用成功,会重回 Closed 状态;
如果第一个调用失败,会进入 Open 状态,然后等待下一次resetTimeout 。
配置
Akka 程序在启动时会加载一些默认的配置项,所以我们不需要显式地提供配置文件,假如要个性化配置某些行为,则需要新建配置文件。在默认情况下,程序会加载 classpath 下的 application.conf 、application.json 、application.properties 文件。
在创建ActorSystem 时,可以显式加载自定义的配置文件,比如app.conf ,代码如下:
ActorSystemsystem=ActorSystem.create("myapp",ConfigFactory.load("app.conf"));
Akka 提供的配置项涵盖了几乎所有的核心功能,比如日志、远程、路由、消息序列化、分布式集群等.