Actor 模型：把并发问题放回业务实体里面

本文是 Agent 工程里的经典架构 系列的一篇。文中的业务场景和代码为讲解设计，产品能力以所链接的官方资料为准。

库存只剩一件，两个处理器几乎同时读到“有货”，随后各自创建了一笔补货或销售任务。每个函数单独看都没有问题，放在一起却产生了冲突。做过共享状态并发处理的人，对这样的故障并不陌生。Agent 能自己决定下一步以后，冲突只会多一种来源，不会自动消失。

Joe Armstrong 的博士论文讨论了利用进程隔离、消息通信和监督机制构建可靠系统。对业务开发者，我认为最容易借鉴的是 Actor 的组织方式：每个实体拥有自己的状态，通过消息与外界交互，把状态的修改收敛到明确的处理入口。

假设我们按商品编号划分库存实体。商品甲和商品乙可以并行处理；商品甲收到的两条扣减消息，则由同一个实体依次判断。第一条扣掉库存后，第二条看到的就是新状态。这里的串行范围只有一个商品，不需要把整个平台所有请求排进同一条全局队列。

下面用 Python 的异步队列模拟一个实体。两个请求并发提交，处理器逐个取出消息，最终只有一笔扣减成功。它只演示消息串行化，不包含 Erlang 的进程隔离、监督树或跨节点寻址，也不是一个持久 Actor 框架。

import asyncio


class StockActor:
    def __init__(self, stock):
        self.stock = stock
        self.mailbox = asyncio.Queue(maxsize=10)

    async def run(self):
        while True:
            message = await self.mailbox.get()
            if message is None:
                return
            quantity, reply = message
            accepted = 0 < quantity <= self.stock
            if accepted:
                self.stock -= quantity
            reply.set_result(accepted)


async def main():
    actor = StockActor(1)
    worker = asyncio.create_task(actor.run())
    replies = [asyncio.get_running_loop().create_future()
               for _ in range(2)]
    for reply in replies:
        await actor.mailbox.put((1, reply))
    result = await asyncio.gather(*replies)
    await actor.mailbox.put(None)
    await worker
    assert result == [True, False]
    assert actor.stock == 0
    print("两条消息顺序处理，库存没有超卖")


asyncio.run(main())

完整示例可下载，使用 Python 3.9 或以上版本运行：python3 example.py。只依赖标准库，不调用真实业务服务。

与“给某段代码加锁”相比，Actor 更强调状态归属。调用方不能随意拿到一个共享字典修改，而是发送业务消息。这样有助于把“谁可以改变库存”变成可以审查的接口设计。不过，如果处理器把可变对象引用泄露给其他线程，这个边界仍然会被绕过；模型的名字不能代替代码约束。

它适合订单、设备、客户会话等天然有身份的对象。对于长期 Agent，可以让客户实体维护稳定联系地址，再为每次服务创建短期任务。回复消息先找到客户，再交给对应任务。需要额外校验任务编号，不能只知道“当前有一个任务”，就把三天前的审批消息交给今天的新申请。

串行化也有代价。热门商品可能形成热点，一个实体的处理速度会成为瓶颈。队列必须有容量上限、超时和拒绝策略，不能把无限排队误认为承受住了流量。慢速外部请求更要小心：如果一次供应商查询堵住处理器十分钟，后面的消息都在等待。可以把外部工作拆出去，再把结果作为新消息送回来，但这时要重新校验状态和版本。

另外，Actor 不天然解决跨实体事务。一次转账涉及两个账户，即使两个账户内部都串行，也仍然需要明确扣款、入账及失败处理的协议。订单与库存之间也是如此。不能因为局部没有竞争，就推导出整个业务流程具备原子性。

故障恢复则是另一层责任。示例的邮箱和库存都在内存中，进程退出后就没有了。监督者能重启进程，但不能凭空恢复业务事实。需要持久状态、事件记录、可重投消息，以及对重复消息的处理。把重启策略和状态恢复策略分别写出来，才能知道系统实际承诺了什么。

消息处理结束到回复调用者之间也可能发生故障。调用者没有收到成功，不代表库存没有扣减；重新发送时仍需识别同一操作。实体内部顺序处理，只消除了同时修改的竞争，不能消除消息重复、响应丢失和网络延迟。

如果需要把实体从一台机器迁移到另一台，还要确保同一时刻只有有效拥有者能写入。仅靠一个稳定地址不够，路由、租约或存储层版本约束需要配合，防止旧实例恢复后继续写入过期状态。

我会在设计时画出实体边界和消息方向，再检查热点、积压、旧消息和跨实体操作。这样使用 Actor，能够把复杂的共享状态拆成更容易理解的局部问题。它是否值得引入，取决于业务是否真的有稳定实体和独立演进的状态，而不取决于我们给对象起了多少个“Agent”的名字。

上一篇 · 系列目录 · 下一篇