---
title: 快速上手Akka
tags:
  - JVM
  - Java
  - Actor
  - 高并发
  - 分布式
  - Akka
  - Spring
categories:
  - - JVM
    - Java
  - - JVM
    - Spring
keywords: 'Java,Akka,Actor,高并发,分布式,并发编程,模式,Spring'
date: 2021-05-20 12:08:22
---

Akka是一个构建分布式高并发，带有容错能力的事件驱动的工具包。Akka有Java和Scala两种实现，是Actor模式的常用实现。在Akka中，Actors是基础执行单元，并且也拥有Actor模式中所涉及的一个Actor要具有的所有元素。<!-- more -->

关于Actor模式可以参考本站文章：{% post_link actor-pattern %}

Akka框架有两个实现，目前在网络中出现最多的是Akka Classic。Akka Classic一般会使用`AbstractActor`或者`UntypedAbstractActor`来定义Actor，这带来的坏处是Actor不能区分所发来消息的类型，也就是Actor所接受的消息不是指定类型的，Actor可以接受任意类型的消息，甚至是`Object`类型。在这种自由类型作为消息的实现模式下，一旦系统规模变大，那么系统中的Actor就会变得难以控制。并且Akka会将Actor无法处理的消息自动转给`unhandled`方法，以保证所有的消息都能得到处理，这也就吞噬了一些潜在异常和逻辑错误。

为了引入一种通讯协议机制，Akka推出了改进以后的Akka Typed实现。明确指定Actor所能够处理的消息类型，为Akka带来了强制通讯协议的支持。所以在使用Akka Typed实现进行系统设计时，就需要采用“协议优先”的设计方法，先从Actor能够处理的消息类型出发开始设计。

虽然Actor模式能够支持数量众多的并发系统设计，但是如果功能粒度切分过细，也同样会引入不必要的复杂度。

## 实现自己的Actor

Actor是Akka框架中的核心元素，也是需要定义的数量最为庞大的内容。在Akka Classic实现中，通常都是使用静态内部类来定义Actor所使用的消息类，但是在Akka Typed实现中，消息类的定义，还是推荐独立出来。

### 定义消息协议

根据“协议优先”的设计原则，要开始定义一个Actor模式的系统，需要首先从Actor之间的交互开始定义。Actor之间用来进行交互的消息主要其实主要有两种：发送和接收，也就是很多示例中常见的Request和Response。

在请求消息的设计上，一般除了会包括消息需要承载的内容以外，通常还会包括消息的发出方。在Akka中，对于其他Actor的引用一般都不直接使用对方Actor的明确类型，而是使用Actor所能够接受的消息类型利用`ActorRef<T>`引用类型来引用。Akka框架的这种设计可以充分的使Actor之间解耦。

!!! warning ""
    在Akka框架中，除了要明确创建子Actor，一般很少会直接使用Actor的类名，都是使用Actor能够接受的消息类名。

为了能够达到通用设计，在设计Actor的消息协议的时候，一般会采用定义空接口作为通用的消息类型，各个Actor中所使用的消息类型均从空接口中派生。这样的设计可以使Actor与消息之间的耦合大大降低。例如Request和Response可以这么设计。

```java
public interface Command {}

public class Request<T> implements Command {
    private T payload;
    private ActorRef<Command> sender;

    // getter和setter先省略了
}

public class Response<T> implements Command {
    private T payload;
    private ActorRef<Command> responder;

    // getter和setter同样先省略
}
```

在网络能够见到还有一种将Request和Response类定义在接口中的写法，这样也是可以的，而且比较节省文件数量（Java会在乎这点儿文件数量么？）。转换成这样的写法如下例所示。

```java
public interface Command {
    class Request<T> implements Command {
        private T payload;
        private ActorRef<Command> sender;

        // getter和setter都已经省略
    }

    class Response<T> implements Command {
        private T payload;
        private ActorRef<Command> responder;

        // getter和setter都已经省略
    }
}
```

!!! note ""
    像上例中这样在接口中定义的类，默认是静态的，实际上就是一个定义在接口里的静态内部类。这样定义的目的主要是将功能用途相近的内容集中放置。

这样定义出来的Request和Response在使用的时候就需要使用`Command.Request<T>`和`Command.Response<T>`的形式了。

在以上两个示例中，可以发现一个很相似的位置就是Request和Response都使用`ActorRef<Command>`引用了消息的发送者。这也是Actor模式在实际使用中的一些习惯，通过`ActorRef<Command>`的定义，Actor可以从消息中获得消息的发送来源。如果Actor的逻辑是需要回复的，那么就可以直接利用消息中附带的发送者引用，直接将所需要发送的数据返回发送者。而在消息体中所有的具名类型，都明确的指示了需要在Actor之间传递什么样的消息。虽然不再像Akka Classic那么自由，但是Actor之间交互所使用的协议变得更加安全了，消息的处理也变得更加明确了。

### Actor Context

Actor Context顾名思义，是一个Actor运行的上下文环境。这个上下文环境通常被用来做以下这些事情。

- 创建子Actor并进行监管。
- 监视其他的Actor收到的终止事件。
- 记录和输出日志。
- 创建消息适配器。
- 利用Ask与其他的Actor进行交互。
- 动态获取Actor自身的引用。

要在Actor中使用Actor Context，可以将其作为Actor构造方法的参数来将其传入。例如常用的Actor的基类`AbstractBehavior`就需要使用`ActorContext`来进行实例化。

### 常用的Actor种类

与Akka Classic中所使用的`AbstractActor`和`AbstractUntypedActor`不同，Akka Typed使用`Behavior<T>`作为Actor的代表。在Akka Typed中，一个`Behavior`就代表了一个处理信息的行为，而在一个`Behavior`中还可以通过返回另一个`Behavior`来表明如何处理下一个消息。

在一般的情况下，要定义一个Actor，大多都是从`AbstractBehavior<T>`类继承来的。除了可以自定义Behavior以外，Akka Typed还通过`Behaviors`类提供了若干拥有固定行为的Actor可供使用。

- `Behaviors.empty()`，会将所有发送来的消息都处理为unhandled。
- `Behaviors.ignore()`，会忽略所有发送来的消息。
- `Behaviors.stopped()`，使当前的Actor终止。
- `Behaviors.unhandled()`，建议系统使用之前的Behavior，包括未处理的行为。
- `Behaviors.same()`，建议系统使用之前的Behavior。

!!! note ""
    从`ExtensibleBehavior<T>`类继承实现Actor是另一种更加原始的选择，但是这样会丢掉`AbstractBehavor<T>`提供的更加便利的操作方法。

### Actor的基本构成

一个Actor完成处理一个消息是通过返回一个新的Behavior来实现的，而Actor里面要实现这一行为最核心的方法就是`ExtensibleBehavior<T>`中提供的`receive()`方法和`reeiveSingal()`方法。这两个方法都需要返回一个`Behavior<T>`实例。

如果Actor是通过继承`AbstractBehavior<T>`来实现的，那么一般会选择实现`createReceive()`方法，这个方法会要求返回一个`Receive<T>`实例，Akka会根据返回的`Receive<T>`实例来创建对应的`receive()`方法和`recceiveSignal()`方法。

在`createReceive()`方法中，一般会使用`AbstractBehavior<T>`类中提供的`newReceiveBuilder()`方法来获取一个`ReceiveBuilder`类的实例，然后再通过对这个实例进行配置来使其构建出我们所需要的`Receive`实例。例如可以参考以下实例。

```java
public class GreetingAction extends AbstractBehavior<GreetingMessage> {

    public GreetingAction(TypedActorContext<GreetingMessage> ctx) {
        super(ctx);
    }

    @Override
    public Receive<GreetingMessage> createReceive() {
        return newReceiveBuilder()
            .onMessage(MorningGreetingMessage.class, message -> {
                // Actor处理消息的过程
                // 发送消息给其他的Actor
                message.getSender().tell(new MorningGreetingMessage());
                // 返回下一个Behavior，this表示当前行为
                return this;
            })
            .build();
    }
}
```

从上面这个例子可以看出来，定义一个Actor的行为，最主要的就是定义`ReceiveBuilder`里的`onMessage()`方法，其实除了`onMessage()`方法以外，`ReceiveBuilder`里还有`onSignal()`方法可供使用。只不过`onSignal()`是用来响应Akka系统信号的。

`ReceiveBuilder`中常用的`onMessage()`方法和`onSignal()`方法的签名有以下几个，可以参考在定义Actor时使用。

- `ReceiveBuilder<T> onAnyMessage(Function<T, Behavior<T>> handler)`，处理所有传入的信息。
- `<M extends T> ReceiveBuilder<T> onMessage(Class<M> type, Function<M, Behavior<T>> handler)`，当传入的信息是`T`的子类型`M`的时候要进行的处理。
- `<M extends T> ReceiveBuilder<T> onMessage(Class<M> type, Predicate<M> test, Function<M, Behavior<T>> handler)`，当传入的信息是`T`的子类型`M`，并同时满足筛选条件的时候要执行的处理。
- `ReceiveBuilder<T> onMessageEquals(T msg, Creator<Behavior<T>> handler)`，当传入信息满足实例相等条件时要执行的Behavior创建过程。
- `<M extends Signal> ReceiveBuilder<T> onSignal(Class<M> type, Function<M, Behavior<T>> handler)`，处理传入的信号是`Signal`的子类型`M`的时候要进行的处理。
- `<M extends Signal> ReceiveBuilder<T> onSignal(Class<M> type, Predicate<M> test, Function<M, Behavior<T>> handler)`，当传入的信号是`Signal`的子类型`M`，并同时满足筛选条件的时候要进行的处理。
- `ReceiveBuilder<T> onSignalEquals(Signal signal, Creator<Behavior<T>> handler)`，当传入信号满足实例相等条件时要执行的Behavior创建过程。

### 生命周期

Actor的生命周期主要包括Actor的创建和停止。在Akka框架中，Actor是有状态的，而且持有相当数量的资源，所以Actor必须被显式创建和停止。

!!! caution ""
    一个Actor在不被引用以后并不会自动停止，所有被显式创建的Actor必须被显式销毁。

!!! caution ""
    如果停止了一个父Actor，那么其下所有被其创建的子Actor都会被停止。同样的，如果停止了Actor System，那么所有的Actor也都会被停止。

要创建一个Actor，可以使用Actor Context提供的`spawn()`方法。`spawn()`方法的方法签名如下`ActorRef<U> spawn(Behavior<U> behavior, String name)`，在`spawn()`方法中，`name`参数只是给创建的Actor起一个名字。

在Actor System中有一个比较特殊的Actor，被称为守护Actor（Guardian Actor）。这个守护Actor除了需要负责其本身的业务逻辑以外，还需要负责创建所有的子系统Actor以及监视他们的生命周期。要创建这个守护Actor不是使用Actor Context提供的`spawn()`方法，而是采用Actor System提供的`create()`方法，这个`create()`方法的方法签名与Actor Context提供的`spawn()`方法的签名一致。

在程序设计中，一般并不推荐直接通过`new`来实例化一个类，所以在Akka里，最常见的创建Actor本身的方法是使用`Behaviors.setup()`方法。其使用可以参考以下示例。

```java
public class GreetingBot extends AbstractBehavior<GreetingMessage> {
    public static Behavior<GreetingMessage> create() {
        return Behaviors.setup(GreetingBot::new);
    }

    private GreetingBot(ActorContext<GreetingMessage> ctx) {
        super(ctx);
    }
}
```

要停止一个Actor并不难，如果一个Actor返回了`Behaviors.stopped()`就会使它自己停止。然后这个Actor就会收到`PostStop`信号，可以允许Actor做一些停止工作以后的清理工作。

如果要监视另一个Actor结束运行的情况，需要监听`Terminated`信号。要监听指定Actor的`Terminated`信号，需要使用Actor Context提供的`watch()`方法来监听指定Actor的活动，如果想计划使用自定义的消息，那么就需要使用`watchWith()`方法了。如果被监视的Actor产生了停止事件，那么监听Actor将会收到`Terminated`信号，可以允许监听Actor做出一些对应的动作。除了可以监听到`Terminated`信号以外，Actor还可以收到其他的信号。

### 消息的传递

Actor之间的消息传递一般都是通过`tell()`方法来完成的，这个方法的使用在前面的示例中已经出现过。`tell()`方法是`ActorRef<T>`提供的，所以如果想要给Actor发送消息，首先就必须先获得这个Actor的引用。

除了可以使用`ActorRef<T>`提供的`tell()`方法以外，还可以借助`ActorContext<T>`提供的`ask()`方法来向其他的Actor传递消息。以下是`ask()`方法的签名信息。

```java
<Req, Res> void ask(
    Class<Res> resClass, 
    RecipientRef<Req> target, 
    Duration timeout, 
    Function<ActorRef<Res>, Req> createRequest, 
    Function2<Res, Throwable, T> applyToResponse
)
```

## Actor System

初看Actor模式的介绍，感觉Actor模式在实现的时候应该所有的Actor都是散在的，或者是都存在于一个Actor池中，相互之间仅通过消息产生关联。但是在Akka框架中，所有的Actor实际上是以一个等级分层结构组织在一起的。所有的Actor根据它们的调用链自然形成了一个具有等级的分层结构。

Actor System是用于存放这个等级分层结构的容器，自然其中的Actor也就按照分层结构被组织在了一起。可以说Actor System是启动整个Actor处理链条的开始。

常用的Actor System实现是位于`akka.actor.typed`包中的`ActorSystem<T>`类。其实阅读过这个类的源码或者API以后就会发现，这个`ActorSystem<T>`类其实也是一个Actor，只不过这个Actor比较特殊，是用来为其他Actor提供运行环境、资源分配和基础设施的。在这个Actor System中，每个Actor都是有其父Actor的，而且从根Actor开始，有三个Actor是比较特殊的。

最根部的Actor一般被称为根节点，通常用`/`表示，它有两个子Actor，一是`/user`，一是`/system`。其中`/system`主要用于Akka系统的内部管理，并不参与用户定义的Actor的运行。用户定义的所有Actor都隶属于`/user`，也都由`/user`负责监管，但是根据Akka的建议，用户自己定义的Actor最好还是由用户自己来进行监管。

在Akka的Actor模式中，Actor的监管机制一般都是采用级联方式处理Actor的失败问题，也就是说父Actor需要决定如何处理子Actor。所以对Actor的分组就影响了监管策略的建立。在默认情况下，Akka Typed采用的策略是停止子Actor，Akka Classic采用的策略是重启子Actor，但是监管窜略如果由我们来制定，那么就可以按照实际需求来定了。

### 子Actor的监管

一个Actor在创建的时候，可以利用`Behaviors.supervise()`方法进行封装，设定其监管策略。例如以下设定代码都是十分常见的。

```java
Behaviors.supervise(behavior).onFailure(SupervisorStrategy.restart());

Behaviors.supervise(behavior).onFailure(IllegalStateException.class, SupervisorStrategy.resume());
```

按照Akka中Actor的组织和监管形式，父级Actor在重新启动的时候，它所创建的所有子Actor也会随之停止和重启。如果想保持子Actor的运行状态，可以参考以下示例代码。

```java
static Behavior<Command> setupParent() {
    return Behaviors.setup(ctx -> {
        final ActorRef<Command> child = ctx.spawn(createChild(), "child1");

        return Behaviors.<Command>supervise(
            Behaviors.receiveMessage(msg -> {
                // 对子Actor进行操作。
                return Behaviors.same();
            })
        )
        .onFailure(SupervisorStrategy.restart().withStopChildren(false));
    });
}
```

!!! caution ""
    如果让`supervise()`包裹`Behaviors.setup()`，那么父级Actor创建的所有子Actor都会随着父级Actor的重启，全部重新创建一遍。所以如果需要保持Actor的运行状态，需要使用`Behaviors.setup()`包裹`supervise()`。

### 获取Actor引用

要获取一个Actor的引用，除了在创建这个Actor的时候保存它的引用以外，还可以使用Actor System提供的Receptionist功能。

Receptionist采用注册制，Actor必须主动在Receptionist上注册才能让其他的Actor通过Receptionist获取它的引用。在Receptionist上注册Actor实际上就是将一个Service Key与Actor建立关联，在Akka中，一个Service Key可以对应多个Actor，这样在向Receptionist请求Actor时，Receptionist会回复一个列表，其中包括所有符合条件的Actor。

要获取一个Receptionist的实例，一般可以通过Actor Context来获取，即`context.getSystem().receptionist()`来获取，或者还可以直接通过`Receptionist`类提供的静态方法来进行操作。

Actor注册用的Service Key是需要通过`ServiceKey.create()`方法来创建的，这个方法的签名为`static <T> ServiceKey<T> create(Class<T> clazz, String id)`。从`create()`方法的签名可以看出来，创建一个Service Key需要提供一个类和一个字符串。

以下是在Receptionist中注册一个Actor的常见过程。

```java
public static Behavior<Command> create() {
    ServiceKey<Command> key = ServiceKey.create(Command.class, "testCommand");
    return Behaviors.setup(context -> {
        context
            .getSystem()
            .receptionist()
            .tell(Receptionist.register(key, context.getSelf()));
        return new ExecuteActor(context);
    })
}
```

在Receptionist中，所有已经注册的Actor组成的注册表是动态的，Actor会根据其状态不同，而从注册表中消失。如果需要监控Receptionist中注册表的变化，可以订阅Receptionist上某个Service Key的变化事件，并将其发送给指定的Actor。订阅事件的编码结构基本上与注册Actor的编码结构相似，只是注册中的`Receptionist.register()`方法需要换成`Receptionist.subscribe()`方法。`Receptionist.subscribe()`方法的签名为`static <T> Receptionist.Command subscribe(ServiceKey<T> key, ActorRef<Receptionist.Listing> subscriber)`。从`Receptionist.subscribe()`方法的签名可以看出，订阅Receptionist变化的Actor需要能够处理`Receptionist.Listing`消息。

在Receptionist中寻找Actor的`Receptionist.find()`方法的签名基本上与`Receptionist.subscribe()`相同。

## 最小入口程序

!!! caution ""
    Actor System是一个会管理运行线程的重量级数据结构，所以原则上一个JVM进程中只创建一个Actor System的实例。

要构建一个最小的入口程序并不难，只需要使用已经定义好的守卫Actor构建Actor System即可，但是不要忘了Actor System也是一个Actor，需要手动关闭和拆解。

```java
public static main(String[] args) {
    final ActorSystem<String> system = ActorSystem.create(Greeting.create(), "Greeting");
    system.tell("start");

    // 这里可以使用Akka Typed提供的Coordinated Shutdown来关闭Actor System并进行一些额外的处理操作
    // 或者还可以使用ActorSystem中的terminated()方法直接进行关闭。
    system.terminated();
}
```

## 与Spring集成

虽然Akka是一个体积不小的框架结构，但是要把它与Spring结合使用，并不是很难。这其中的关键是要明确Akka中的哪些内容是可以有Spring来管理的。

Spring中最底层的功能是IoC，这需要Spring能够掌握系统中绝代多数的组件实例才可以。但是在Akka中，Actor都是有Actor System控制的，并且都是呈登记分层结构组织的，而且Actor的实例基本上都是在父级Actor中创建的。明白了这些内容，就基本上明白了将Akka与Spring结合时所需要解决的难点。

首先必须要实现的是，把Actor System实例托管给Spring。这一步非常简单，只需要在Spring中实例化一个单例的Bean即可。其次，对于Actor的生成，可以利用Spring的`prototype`级别的Bean来生成。例如以下示例。

```java
@Named("greeting_bot")
@scope("prototype")
public class GreetingBot extends AbsractBehavior<Greeting> {
    // 省略Actor中具体的实现
}
```

使用`@Named`注解主要是因为Akka在初始化Actor的时候必须使用Actor System或者Actor Context提供的工厂方法，不能直接使用构造函数。

另一种生成Actor的方法是实现Akka中提供的`IndirectActorProducer`接口。通过这个接口可以定制一些Actor的生成方式。例如可以构建这样一个通过构造函数生成Actor的Producer。

```java
public class SpringActorProducer implements IndirectActorProducer {
    private final ApplicationContext applicationContext;
    private final String actorBeanName;
    private final Object[] args;

    public SpringActorProducer(ApplicationContext applicationContext, String actorBeanName, Object... args) {
        this.applicationContext = applicationContext;
        this.actorBeanName = actorBeanName;
        this.args = args;
    }

    public Behavior produce() {
        return (Behavior)applicationContext.getBean(actorBeanName, args);
    }

    public Class<? extends Behavior> actorClass() {
        return (Class<? extends Behavior>)applicationContext.getType(actorBeanName);
    }
}
```

之后可利用其构建一个用于生成Actor的代理。

```java
@Component("spring_ext")
public class SpringExtension implements Extension, ApplicationContextAware {
    private ApplicationContext applicationContext;

    public Props props(String actorBeanName, Object... args) {
        return Props.create(SpringActorProducer.class, applicationContext, actorBeanName, args);
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
```

!!! caution "请勿直接使用的警告"
    以上Akka与Spring结合的代码转摘自Akka Classic并做了简单的对应到Akka Typed的处理，尚未在实际代码中做验证。如需在项目中使用，可先进行测试验证。