Akka Actor编程

Akka actor 模型是一种强大的并发编程范式，它通过消息传递来实现 actor 之间的通信，避免了传统并发编程中的锁和共享状态的问题。

以下是 Akka actor 编程的一个基本教程。

1. 基础概念

在 Akka 中，一切都是 actor。actor 是 Akka 并发模型的基本构建块，它封装了状态和行为，并通过消息传递与外界通信。每个 actor 都有一个邮箱（mailbox），用于接收消息，以及一个行为（behavior），定义了如何处理这些消息。

2. 创建 ActorSystem

ActorSystem 是创建和管理 actor 的根级别的容器。在 Java 中，你可以这样创建一个 ActorSystem：

1
2
3

import akka.actor.ActorSystem;

ActorSystem system = ActorSystem.create("mySystem");

3. 定义 Actor

在 Akka 中，所有的 actor 都是通过扩展 AbstractActor 类或使用 AbstractBehavior 来创建的。下面是一个简单的 actor 示例，它接收字符串消息并打印出来：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import akka.actor.AbstractActor;
import akka.actor.Props;
import akka.actor.ActorRef;
import akka.japi.pf.ReceiveBuilder;

public class MyActor extends AbstractActor {
    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(String.class, message -> {
                System.out.println("Received message: " + message);
            })
            .build();
    }
}

// 创建 Props 对象，它包含了 actor 的配置信息
Props props = Props.create(MyActor.class);

// 使用 ActorSystem 创建一个 actor 实例
ActorRef myActorRef = system.actorOf(props, "myActor");

4. 发送消息

在 Akka 中，你可以通过 tell 方法向 actor 发送消息。这个方法不关心消息是否被处理，它只是将消息发送出去：

1

myActorRef.tell("Hello, Akka!", ActorRef.noSender());

5. 处理消息

在上面的 MyActor 示例中，我们定义了一个 Receive 构建器，它匹配传入的 String 类型消息，并打印出来。你可以根据需要定义更多的消息类型和处理逻辑。

6. 停止 Actor

Akka actor 可以通过 getContext().stop(getSelf()); 来停止自己，或者通过 getContext().stop(targetActorRef); 来停止其他 actor。

7. 监督和监控

Akka 允许你设置 actor 的监督策略，以决定当子 actor 失败时如何处理。你可以定义一个 SupervisorStrategy 来决定是重启、停止还是忽略失败。

8. 持久化和分片

Akka 还支持 actor 的持久化和分片，这对于构建大规模分布式系统非常有用。通过持久化，你可以确保 actor 的状态在系统崩溃后能够恢复。分片则允许你将 actor 分布到多个节点上，以提高系统的可伸缩性。

以上是 Akka actor 编程的一个基本介绍。Akka 是一个功能丰富的框架，它提供了许多高级特性，如分布式系统支持、流处理等。你可以在 Akka 官方文档中找到更多详细的信息和高级教程 。

Akka actor 之间的通信通常是基于消息传递的。这里提供一个简单的 Akka actor 通信模式示例，包括一个父 actor 和两个子 actor。父 actor 将消息分发给两个子 actor，并处理它们的响应。

首先，我们需要创建一个简单的消息类来表示 actor 之间的通信内容：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// 消息类
public class Greeting {
    private String name;

    public Greeting(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }
}

接下来，我们定义两个子 actor，Greeter 和 Responder：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import akka.actor.AbstractActor;
import akka.actor.Props;
import akka.actor.ActorRef;

// Greeter Actor
public class Greeter extends AbstractActor {
    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(Greeting.class, greeting -> {
                String reply = "Hello " + greeting.getName();
                getSender().tell(reply, getSelf());
            })
            .build();
    }
}

// Responder Actor
public class Responder extends AbstractActor {
    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .matchAny(message -> {
                getSender().tell(message, getSelf());
            })
            .build();
    }
}

然后，我们创建一个父 actor，Supervisor，它将消息发送给 Greeter 和 Responder：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import akka.actor.AbstractActor;
import akka.actor.Props;
import akka.actor.ActorRef;

public class Supervisor extends AbstractActor {
    public Supervisor() {
        // 创建子 actor
        ActorRef greeter = getContext().actorOf(Props.create(Greeter.class), "greeter");
        ActorRef responder = getContext().actorOf(Props.create(Responder.class), "responder");

        // 将子 actor 引用保存为路径，以便发送消息
        getContext().watch(greeter); // 监视 greeter 的生命周期
        getContext().watch(responder); // 监视 responder 的生命周期
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .match(Greeting.class, greeting -> {
                ActorRef greeter = getContext().actorOf(Props.create(Greeter.class), "greeter");
                ActorRef responder = getContext().actorOf(Props.create(Responder.class), "responder");

                // 将消息发送给 greeter 和 responder
                greeter.tell(greeting, getSelf());
                responder.tell(greeting, getSelf());
            })
            .match(String.class, message -> {
                // 处理 greeter 和 responder 的响应
                System.out.println(message);
            })
            .build();
    }
}

最后，我们创建一个 ActorSystem 并启动这个 actor 树：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import akka.actor.ActorSystem;

public class ActorCommunicationExample {
    public static void main(String[] args) {
        // 创建 ActorSystem
        ActorSystem system = ActorSystem.create("ActorExample");

        // 创建 Supervisor actor
        ActorRef supervisor = system.actorOf(Props.create(Supervisor.class), "supervisor");

        // 发送消息给 Supervisor
        supervisor.tell(new Greeting("Akka"), ActorRef.noSender());
    }
}

在这个示例中，我们创建了一个 Greeting 类型的消息，两个子 actor Greeter 和 Responder，以及一个父 actor Supervisor。Supervisor 接收 Greeting 类型的消息，并将它们分发给 Greeter 和 Responder。Greeter 回复一个问候消息，而 Responder 简单地将消息转发回 Supervisor。Supervisor 然后打印出这些响应。

这个示例展示了 Akka actor 之间如何通过消息传递进行通信。每个 actor 都是一个并发的实体，它们通过消息传递来交互，而不是共享内存。这种模型有助于编写无锁的并发代码，从而提高应用程序的可伸缩性和容错性。

在 Akka actor 模型中，有几个常用的类和接口，它们构成了 Akka 应用程序的基础。以下是一些关键的类和接口的详细说明：

  1. ActorSystem：

    • ActorSystem 是 Akka 应用程序的入口点，负责创建和管理 actor 的生命周期、线程池、消息队列等。它是重量级对象，通常一个应用程序中只有一个 ActorSystem 实例。

  2. ActorRef：

    • ActorRef 是对 actor 的引用，可以看作是 actor 的代理。每个 actor 都有一个唯一的 ActorRef，通过它来发送消息和进行远程调用。

  3. Props：

    • Props 类包含了 actor 的配置信息，如调度器、邮箱或部署配置。它用于创建新的 actor 实例。Props 是不可变的，因此是线程安全的。

  4. AbstractActor：

    • 这是定义 actor 行为的基础抽象类。通过扩展这个类，可以定义 actor 的行为，如接收和处理消息。

  5. ActorContext：

    • ActorContext 提供了关于 actor 的上下文信息，如 actor 的路径、父 actor 引用等。它通常在 actor 的构造函数中作为参数传入，并用于访问 actor 的环境信息。

  6. ReceiveBuilder：

    • ReceiveBuilder 是用于定义 actor 行为的工具，它允许你以声明式的方式定义消息处理逻辑。

  7. ActorSelection：

    • ActorSelection 用于选择一个 actor，即使它的 ActorRef 不是直接已知的。它允许你向一个 actor 路径发送消息，而不需要直接引用该 actor 的 ActorRef。

  8. SupervisorStrategy：

    • 监管策略定义了 actor 遇到异常时的处理方式。你可以定义自己的监管策略来决定是重启、停止还是忽略子 actor 的失败。

  9. TypedActor：

    • TypedActor 是 Scala 中的一个特性，它允许你以类型安全的方式定义 actor。

  10. ActorPath：

    • ActorPath 表示 actor 在 actor 系统中的位置。每个 actor 都有一个唯一的路径，可以用来唯一标识一个 actor。

  11. Dispatcher：

    • 调度器负责管理 actor 执行的线程和调度执行顺序。Akka 允许你配置不同的调度器，以适应不同的并发和性能需求。

  12. Mailbox：

    • 消息队列用于存储 actor 接收到的消息，并根据调度器的调度顺序将消息发送给 actor 执行。

  13. ActorOf：

    • actorOf 是一个工厂方法，用于创建新的 actor 实例。它接受 Props 作为参数，并返回一个新的 ActorRef。

  14. tell() 和 ask()：

    • tell() 方法用于向 actor 发送消息，它是异步的，不会等待接收方 actor 处理完消息。
    • ask() 方法用于发送请求并期望得到响应，它返回一个 Future 对象，可以用于异步组合。

这些类和接口共同构成了 Akka actor 模型的基础，使得开发者可以构建高度并发和分布式的应用程序。通过这些组件，Akka 提供了强大的工具来处理并发、状态管理和消息传递。