Akka Typed 实战指南：从 Actor 模型到集群分片的完整架构设计

Akka Typed 实战指南：从 Actor 模型到集群分片的完整架构设计

在 Scala 生态中，Akka 是构建高并发、分布式、弹性消息驱动应用的事实标准框架。2021年发布的 Akka Typed（akka-actor-typed）彻底改变了传统经典 Akka（Classic Actors）的 API 设计，将 Actor 的行为建模为类型安全的函数式组合，使得编译期就能捕获大量 Actor 交互错误。

本文将从零开始，系统讲解 Akka Typed 的核心概念、行为组合模式、持久化事件溯源、测试策略，以及生产级集群部署（Cluster Sharding）。所有代码均基于 Akka 2.9.x 版本，使用 Scala 3 语法编写。

一、Akka Typed 核心概念

1.1 行为（Behavior）是 Actor 的灵魂

在 Akka Classic 中，Actor 通过继承

特质并实现

方法来定义行为。Akka Typed 彻底改变了这一设计：Actor 不再是一个类，而是一个 行为函数

，其中

是该 Actor 能处理的消息类型。

1
2
3
4
5
6
7
8
9
10
// 最简单的 Behavior：接收 String 消息

import akka.actor.typed.Behavior

import akka.actor.typed.scaladsl.Behaviors



object Greeter {

  def apply(): Behavior[String] = Behaviors.receiveMessage { name =>

    println(s"Hello, $name!")

    Behaviors.same  // 保持当前行为

  }

}

关键设计原则：

• 类型安全：Actor 只能接收声明类型的消息，编译器保证
• 不可变行为：

  1

  Behavior

  是不可变值，通过组合产生新行为

• 显式状态转换：状态变化通过返回新的

  1

  Behavior

  来表达

1.2 ActorSystem 与 ActorRef

启动 Actor 系统并创建 Actor 实例：

1
2
3
4
5
6
import akka.actor.typed.ActorSystem



val system: ActorSystem[String] = ActorSystem(Greeter(), "hello-system")

system ! "World"    // 发送消息

system ! "Akka"     // 输出: Hello, Akka!

system.terminate()  // 关闭系统

是 Actor 的引用指针，类型参数

声明了该 Actor 能接收的消息类型。使用

（

操作符）发送消息是异步的、非阻塞的。

二、状态管理与行为切换

2.1 可变状态 vs 行为切换

Akka Typed 提供了两种状态管理模式：

推荐使用行为切换模式，因为它的状态转换是显式的，便于测试和推理：

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
34
35
36
37
38
39
40
// 有限状态机：投票系统

sealed trait VoteCommand

case class Vote(candidate: String, replyTo: akka.actor.typed.ActorRef[VoteResult]) extends VoteCommand

case class GetResult(replyTo: akka.actor.typed.ActorRef[VoteResult]) extends VoteCommand

case class VoteResult(candidate: String, count: Int)



object Ballot {

  def apply(): Behavior[VoteCommand] = Behaviors.setup { _ =>

    idle(Map.empty)

  }



  private def idle(votes: Map[String, Int]): Behavior[VoteCommand] =

    Behaviors.receiveMessage {

      case Vote(candidate, replyTo) =>

        val updated = votes.updated(candidate, votes.getOrElse(candidate, 0) + 1)

        replyTo ! VoteResult(candidate, updated(candidate))

        active(updated)  // 切换到 active 状态

      case GetResult(replyTo) =>

        replyTo ! VoteResult("open", 0)

        Behaviors.same

    }



  private def active(votes: Map[String, Int]): Behavior[VoteCommand] =

    Behaviors.receiveMessage {

      case Vote(candidate, replyTo) =>

        val updated = votes.updated(candidate, votes.getOrElse(candidate, 0) + 1)

        replyTo ! VoteResult(candidate, updated(candidate))

        if (updated.values.sum >= 100) closed(updated)  // 达到阈值→关闭

        else active(updated)

      case GetResult(replyTo) =>

        votes.foreach { case (c, n) => replyTo ! VoteResult(c, n) }

        Behaviors.same

    }



  private def closed(votes: Map[String, Int]): Behavior[VoteCommand] =

    Behaviors.receiveMessage {

      case _ =>

        Behaviors.unhandled  // 投票已关闭，拒绝所有消息

    }

}

三、Actor 通信模式

3.1 Ask 模式：请求-响应

从外部向 Actor 发送请求并等待响应，使用

模式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import akka.actor.typed.scaladsl.AskPattern._

import scala.concurrent.duration._

import scala.util.Success



implicit val system: ActorSystem[VoteCommand] = ???

implicit val timeout: akka.util.Timeout = 5.seconds



val ballot: akka.actor.typed.ActorRef[VoteCommand] = system ! Ballot()



val result: Future[VoteResult] = ballot.ask(ref => GetResult(ref))

result.onComplete {

  case Success(VoteResult(c, n)) => println(s"$c: $n votes")

  case _ => println("Query failed")

}

返回一个

，需要隐式的

和

。实际项目中建议使用

或自定义错误类型来避免 Future 的异常处理陷阱。

3.2 Adapt 模式：消息适配

当一个 Actor 需要与多个不同类型的 Actor 通信时，使用消息适配器统一消息类型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 父 Actor 管理子 Actor

sealed trait ParentCommand

case class StartChild(name: String) extends ParentCommand

private case class ChildResponse(msg: String) extends ParentCommand



object Parent {

  def apply(): Behavior[ParentCommand] = Behaviors.setup { ctx =>

    Behaviors.receiveMessage {

      case StartChild(name) =>

        val child: ActorRef[String] = ctx.spawn(Greeter(), name)

        // 将子 Actor 的 String 响应适配为 ParentCommand

        val adapter: ActorRef[String] = ctx.messageAdapter(msg => ChildResponse(msg))

        child ! s"Hello from parent"

        Behaviors.same

      case ChildResponse(msg) =>

        println(s"Received from child: $msg")

        Behaviors.same

    }

  }

}

四、持久化与事件溯源

4.1 EventSourcedBehavior 基础

Akka Persistence Typed 将事件溯源（Event Sourcing）作为一等公民：

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
34
35
36
37
38
39
40
41
import akka.persistence.typed.scaladsl.EventSourcedBehavior

import akka.persistence.typed.PersistenceId



// 事件

sealed trait AccountEvent

case class Deposited(amount: BigDecimal, timestamp: Long) extends AccountEvent

case class Withdrawn(amount: BigDecimal, timestamp: Long) extends AccountEvent



// 状态

case class AccountState(balance: BigDecimal, lastUpdated: Long)



object Account {

  def apply(accountId: String): Behavior[AccountCommand] =

    EventSourcedBehavior[AccountCommand, AccountEvent, AccountState](

      persistenceId = PersistenceId.ofUniqueId(accountId),

      emptyState = AccountState(BigDecimal(0), 0L),

      commandHandler = (state, cmd) => handleCommand(state, cmd),

      eventHandler = (state, evt) => handleEvent(state, evt)

    )



  private def handleCommand(state: AccountState, cmd: AccountCommand): Effect[AccountEvent, AccountState] =

    cmd match {

      case Deposit(amount, _) =>

        Effect.persist(Deposited(amount, System.currentTimeMillis()))

      case Withdraw(amount, replyTo) if state.balance >= amount =>

        Effect.persist(Withdrawn(amount, System.currentTimeMillis()))

          .thenReply(replyTo)(_ => WithdrawSuccess)

      case Withdraw(_, replyTo) =>

        Effect.reply(replyTo)(InsufficientBalance(state.balance))

      case GetBalance(replyTo) =>

        Effect.reply(replyTo)(CurrentBalance(state.balance))

    }



  private def handleEvent(state: AccountState, evt: AccountEvent): AccountState =

    evt match {

      case Deposited(amount, ts) =>

        state.copy(balance = state.balance + amount, lastUpdated = ts)

      case Withdrawn(amount, ts) =>

        state.copy(balance = state.balance - amount, lastUpdated = ts)

    }

}

事件溯源的核心优势：

• 完整审计日志：所有状态变更以事件形式持久化，可追溯
• 时间旅行：通过重放事件重建任意时刻的状态
• 读写分离：命令处理（写）与状态查询（读）可独立扩展

4.2 快照与恢复性能

当事件数量达到数十万级别时，从零重放性能不可接受。Akka 提供了快照机制：

1
2
3
4
5
6
7
8
9
EventSourcedBehavior(

  persistenceId = PersistenceId.ofUniqueId(accountId),

  emptyState = AccountState(BigDecimal(0), 0L),

  commandHandler = handleCommand,

  eventHandler = handleEvent

).withSnapshotPlugin("akka.persistence.snapshot-store.local")

 .withRetention(

   RetentionCriteria(snapshotEveryNEvents = 100, keepNSnapshots = 2)

 )

每 100 个事件自动保存快照，重启时从最近的快照开始恢复，只重放快照之后的事件，恢复时间从 O(N) 降到 O(1)。

五、测试策略

5.1 ActorTestKit 同步测试

Akka Typed 提供了

进行行为测试：

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
import akka.actor.testkit.typed.scaladsl.ActorTestKit

import org.scalatest.BeforeAndAfterAll

import org.scalatest.wordspec.AnyWordSpec



class BallotSpec extends AnyWordSpec with BeforeAndAfterAll {

  val testKit = ActorTestKit()



  "Ballot Actor" should {

    "record votes correctly" in {

      val probe = testKit.createTestProbe[VoteResult]()

      val ballot = testKit.spawn(Ballot(), "test-ballot")



      ballot ! Vote("Alice", probe.ref)

      probe.expectMessage(VoteResult("Alice", 1))



      ballot ! Vote("Alice", probe.ref)

      probe.expectMessage(VoteResult("Alice", 2))



      ballot ! Vote("Bob", probe.ref)

      probe.expectMessage(VoteResult("Bob", 1))

    }



    "reject votes after closure" in {

      // 通过发送大量消息触发 closure

    }

  }



  override def afterAll(): Unit = testKit.shutdownTestKit()

}

是同步测试的基石，支持

、

、

等断言方法，默认超时 3 秒。

5.2 持久化 Actor 测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import akka.persistence.testkit.scaladsl.EventSourcedBehaviorTestKit



class AccountPersistenceSpec extends AnyWordSpec {

  val eventSourcedTestKit =

    EventSourcedBehaviorTestKit[AccountCommand, AccountEvent, AccountState](

      system, Account("test-account"))



  "Account" should {

    "persist deposit events" in {

      val result = eventSourcedTestKit.runCommand(Deposit(100, ref))

      result.state.balance shouldBe BigDecimal(100)

      result.events shouldBe List(Deposited(100, _))

    }



    "reject insufficient withdrawal" in {

      val result = eventSourcedTestKit.runCommand(Withdraw(200, ref))

      result.reply shouldBe InsufficientBalance(BigDecimal(100))

      result.hasNoEvents shouldBe true

    }

  }

}

在内存中模拟持久化存储，无需启动真实的数据库，测试速度极快（毫秒级）。

六、集群部署：Cluster Sharding

6.1 Sharding 架构原理

当单个 Actor 实例无法承载负载时，需要将 Actor 分散到集群节点。Akka Cluster Sharding 使用 一致性哈希 将实体（Entity）分配到分片（Shard），分片再分配到集群节点：

1
Entity ID → Shard ID (hash mod N) → Cluster Node

配置示例（application.conf）：

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

  actor.provider = cluster



  remote.artery {

    canonical.hostname = "node1.example.com"

    canonical.port = 25520

  }



  cluster {

    seed-nodes = [

      "akka://accounts@node1.example.com:25520",

      "akka://accounts@node2.example.com:25520"

    ]

  }



  cluster.sharding {

    number-of-shards = 100

    remember-entities = on

    entity-restart-timeout = 30s

    passivate-idle-entity-after = 5.minutes

  }

}

6.2 启动 Cluster Sharding

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import akka.cluster.sharding.typed.scaladsl.{ClusterSharding, Entity}

import akka.cluster.sharding.typed.ShardingEnvelope



val sharding = ClusterSharding(system)



val accountShard: ActorRef[ShardingEnvelope[AccountCommand]] =

  sharding.init(Entity(

    typeKey = EntityTypeKey[AccountCommand]("Account"),

    createBehavior = entityContext =>

      Account(entityContext.entityId)

  ).withStopMessage(PassivateAccount))



// 通过 ShardingEnvelope 发送消息

accountShard ! ShardingEnvelope("acc-001", Deposit(1000))

accountShard ! ShardingEnvelope("acc-002", GetBalance(replyTo))

6.3 分片分配策略

默认使用

，将新分片分配到节点数量最少的节点。对于需要自定义分布的场景：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
Entity(/* ... */)

  .withAllocationStrategy(

    new ShardAllocationStrategy {

      override def allocateShards(

        requester: ActorRef[ShardAllocationStrategy.LeastShardAllocationState],

        shardIds: Set[String],

        rebalance: Set[String]

      ): Future[Map[String, IndexedSeq[String]]] = {

        // 自定义分片分配逻辑

        ???

      }

      override def rebalance(

        currentShardAllocations: Map[String, IndexedSeq[String]],

        rebalance: Set[String]

      ): Future[Set[String]] = {

        // 自定义再平衡策略

        ???

      }

    }

  )

七、生产级最佳实践

7.1 监督与容错

1
2
3
Behaviors.supervise[AccountCommand](Account(accountId))

  .onFailure[IllegalStateException](SupervisorStrategy.restart.withStopChildren(false))

  .onFailure[Exception](SupervisorStrategy.restart.withLimit(maxNrOfRetries = 3, withinTimeRange = 10.seconds))

7.2 性能调优建议

7.3 常见陷阱

• Ask 超时导致 Future 泄漏：始终为

  1

  ask

  设置合理的超时时间，并在超时后清理资源

• 消息过大：Artery 默认帧大小 256 KiB，超过此限制的消息会被静默丢弃。使用

  1	akka.remote.artery.advanced.maximum-frame-size

  调整

• 分片热点：Entity ID 分布不均匀会导致某些分片承载过多实体。使用

  1	number-of-shards = 10 * cluster_size

  的经验公式

• 事件数量爆炸：未设置快照策略时，重启恢复时间随事件数线性增长。务必配置

  1	withRetention

• 阻塞操作：在 Actor 中执行阻塞 I/O 操作会耗尽调度线程。使用

  1	Behaviors.blockingExecutor

  或在单独的

  1	Dispatcher

  中运行

八、从 Akka Classic 迁移指南

Akka Typed 与 Classic 不兼容，但 Akka 提供了

包实现双向适配：

1
2
3
4
5
6
7
8
9
10
import akka.actor.typed.scaladsl.adapter._



// Classic ActorSystem → Typed ActorSystem

val typedSystem: ActorSystem[Nothing] = classicSystem.toTyped



// Typed ActorRef → Classic ActorRef

val classicRef: akka.actor.ActorRef = typedRef.toClassic



// Classic Props → Typed Behavior（包裹）

val typedBehavior: Behavior[Any] = classicProps.toTyped

建议策略：

1. 新模块直接使用 Akka Typed API
2. 旧模块通过适配器逐步迁移，每次迁移一个 Actor
3. 优先迁移无状态或纯计算型 Actor，积累经验后再迁移有状态 Actor
4. 使用

   1	-Xlint:deprecation

   编译选项识别 Classic API 使用

总结

Akka Typed 通过类型安全的 Behavior API、事件溯源的一等公民支持、以及 Cluster Sharding 的声明式配置，将并发编程的复杂度大幅降低。本文涉及的 Actor 模型、持久化、测试和集群部署构成了生产级 Akka 应用的完整技术栈。

下一步深入学习方向：

• Akka Projections：构建事件驱动 CQRS 读模型
• Akka HTTP + Akka gRPC：构建微服务通信层
• Alpakka：连接 Kafka、Kinesis、Pulsar 等消息队列
• Akka Management + Akka Diagnostic：集群监控与排障

希望本文能帮助你快速上手 Akka Typed，构建稳健的 Scala 并发系统。