Agent范式解析：从基础概念到分布式系统实践

1次阅读

共计 2567 个字符，预计需要花费 7 分钟才能阅读完成。

Agent 范式是一种将计算实体建模为独立自治单元（Agent）的编程模型。每个 Agent 拥有自己的状态和行为，通过异步消息传递进行通信。与 OOP 的核心差异体现在：

状态管理：OOP 通过对象属性暴露状态，Agent 强制封装状态（内部可变性）
通信机制：OOP 使用同步方法调用，Agent 采用异步消息传递
并发模型：OOP 依赖共享内存 + 锁，Agent 实现无共享架构

状态封装：每个 Agent 维护私有状态，外部仅能通过消息修改状态，天然避免竞态条件
消息驱动：基于事件的处理模式适合分布式环境，消息队列提供天然缓冲
容错机制：采用『let-it-crash』哲学，通过监督树实现层级容错

import akka.actor.{ActorSystem, Props}

val system = ActorSystem("BankingSystem")
// 创建账户 Actor 集群
val accountA = system.actorOf(Props[BankAccount], "accountA")
val accountB = system.actorOf(Props[BankAccount], "accountB")

sealed trait AccountCommand
case class Deposit(amount: BigDecimal) extends AccountCommand
case class Withdraw(amount: BigDecimal) extends AccountCommand
case class Transfer(to: ActorRef, amount: BigDecimal) extends AccountCommand

sealed trait AccountEvent
case class BalanceUpdated(newBalance: BigDecimal) extends AccountEvent
case class InsufficientFunds(current: BigDecimal) extends AccountEvent

class BankAccount extends Actor {private var balance = BigDecimal(0)

  def receive = {case Deposit(amount) =>
      balance += amount
      sender() ! BalanceUpdated(balance)

    case Transfer(to, amount) if balance >= amount =>
      balance -= amount
      to ! Deposit(amount)
      sender() ! BalanceUpdated(balance)

    case _: Transfer =>
      sender() ! InsufficientFunds(balance)
  }
}

class AccountSupervisor extends Actor {override val supervisorStrategy = OneForOneStrategy() {
    case _: ArithmeticException => Resume  // 继续处理下条消息
    case _: IllegalArgumentException => Restart // 重建 Actor
    case _ => Escalate  // 交由上级处理
  }

  def receive = {
    case cmd: AccountCommand => 
      context.child("bankAccount").getOrElse(context.actorOf(Props[BankAccount], "bankAccount")
      ) forward cmd
  }
}

邮箱类型	吞吐量(msg/s)	延迟(99% 分位)
无界队列	1,200,000	15ms
有界队列(10000)	850,000	8ms
优先级邮箱	600,000	22ms

推荐策略：
– 高吞吐场景使用无界邮箱
– 低延迟场景使用有界邮箱 + 背压

瞬时故障：Resume策略保持状态
可恢复错误：Restart重建内部状态
致命错误：Stop防止状态污染

监控 mailbox-size 指标
使用 DeadLetter 监听器检测未送达消息
配置 akka.actor.debug.receive 日志级别

system.eventStream.subscribe(deadLetterListener, classOf[DeadLetter])

class DeadLetterListener extends Actor {
  def receive = {
    case d: DeadLetter => 
      log.warning(s"Detected dead letter: ${d.message}")
      // 可加入重试或报警逻辑
  }
}

ClusterSharding(system).start(
  typeName = "Accounts",
  entityProps = Props[BankAccount],
  settings = ClusterShardingSettings(system),
  extractEntityId = {case cmd: AccountCommand => (cmd.accountId.toString, cmd)
  },
  extractShardId = {
    case cmd: AccountCommand => 
      (cmd.accountId.hashCode % 100).abs.toString
  }
)

在 Actor 粒度设计中需要考虑：
– 细粒度 Actor（如每个订单一个 Actor）提升并行度但增加调度开销
– 粗粒度 Actor（如每个用户一个 Actor）减少通信成本但降低并发性

建议通过以下维度权衡：
1. 业务实体生命周期
2. 状态更新频率
3. 消息传递的热点分布
4. 硬件资源限制

本文展示的完整示例代码已上传 Github（需替换为真实链接），读者可基于该实现进行扩展实验。在实际业务中采用 Agent 范式时，建议从简单用例开始逐步验证模型合理性，再向复杂场景演进。

正文完