VOL.III · 心跳CH 10监护 10/13

并发工具箱:Task、Agent 与 ETS

GenServer 是心脏,但不是每个器官都得是心脏。VOL.III 的收官章清点 OTP 工具箱里剩下的常备件:外包跑腿的 Task、极简存物的 Agent、高速共享读的 ETS,以及支撑「每个订单一个进程」的 Registry + DynamicSupervisor 组合。学完本章,第 8 章欠的三张药方全部兑现,你手里将有一张完整的并发选型表。

TaskAgentETSDynamicSupervisor

Task:把慢活外包出去

第 8 章的病例:回调里做 3 秒的 HTTP 调用,整条队列陪葬。药方是外包——而「spawn 出去干活、干完把结果寄回来、超时和崩溃都有人管」这套流程的标准件就是 Task:

iex(1)> task = Task.async(fn -> :timer.sleep(2000); {:ok, "报告出炉"} end)
%Task{...}
iex(2)> # 这 2 秒里当前进程可以继续干别的
iex(3)> Task.await(task, 5000)
{:ok, "报告出炉"}

async 立刻返回一个任务凭据,await 凭据取结果(默认限时 5 秒,超时按第 6 章的哲学处理:崩,别无限等)。机制上它就是 spawn + monitor + receive 的封装——第 7 章你手写过的乒乓,标准化了。真正的大杀器是集合版:

bed_ids
|> Task.async_stream(&fetch_vitals/1, max_concurrency: 10, timeout: 5_000)
|> Enum.map(fn {:ok, result} -> result end)

一百个床位的数据要拉取:串行太慢,全并发会打挂下游。async_stream 是「并发版的 Enum.map」:每个元素一个进程,但同时在跑的最多 max_concurrency 个,结果按输入顺序吐出。限并发、限超时、顺序保持,三件运维最操心的事变成三个参数。这一个函数,在别的技术栈里是一个线程池 + 信号量 + Future 编排的小工程。

一个必须划重点的机制细节:Task.async 会把任务进程和你 link 起来(第 9 章的生死契约)——任务崩了,你也崩。这是刻意的:「我在等它的结果,它死了我拿不到,不如一起交给监督树处理」。但如果你想要「它崩它的,别连累我」,用 Task.Supervisor.async_nolink:任务挂在专门的监督者下,和你只有 monitor 关系,死讯以 {:DOWN, ...} 消息温和送达。选 async 还是 async_nolink,就是在选「共命运还是旁观」——link 与 monitor 的那道选择题,在工具箱层面原样重现。

不需要结果的火忘型(fire-and-forget)任务,用 Task.Supervisor.start_child 挂到专门的监督者下面跑——上一章的纪律别忘:不许有无人监护的进程,哪怕是跑腿的。

Agent:只存东西,不办业务

有时你要的不是「服务」,只是「一个能被多方读写的值」——一份运行时配置、一张小缓存表、测试里的假状态。为此写全套 GenServer 六个回调,像为了存一把伞盖一间房。Agent 是那把伞架:

iex(1)> {:ok, _} = Agent.start_link(fn -> %{} end, name: Config)
iex(2)> Agent.update(Config, &Map.put(&1, :ward, "急诊"))
:ok
iex(3)> Agent.get(Config, & &1.ward)
"急诊"

get 传一个「怎么读」的函数,update 传一个「怎么改」的函数,完事。它内部就是个 GenServer(状态即那个值,回调即你传的函数),所以单进程串行的性能模型、可被监督、可被 :sys.get_state 体检,全部照旧。使用边界也因此清晰:状态一旦长出业务逻辑(校验、副作用、多步流程),立刻升级成正经 GenServer——把业务函数塞进 Agent.update 里,是把逻辑藏进匿名函数的坏味道。

ETS:绕开串行瓶颈的共享内存表

第 8 章的第二张药方,也是 BEAM 老资历的性能王牌。场景:配置缓存,一秒被读一万次,写每分钟一次。走 GenServer,一万次读排一条队——纯粹的自我惩罚。BEAM 给了一个「官方后门」:ETS(Erlang Term Storage),虚拟机内建的内存键值表:

iex(1)> :ets.new(:vitals_cache, [:set, :named_table, :public, read_concurrency: true])
:vitals_cache
iex(2)> :ets.insert(:vitals_cache, {:bed_7, %{hr: 72, spo2: 98}})
true
iex(3)> :ets.lookup(:vitals_cache, :bed_7)
[bed_7: %{hr: 72, spo2: 98}]

建表选项先认两组:表类型 :set(键唯一,最常用)/:ordered_set(按键有序,能范围扫描)/:bag(一键多值);可见性 :public(谁都能读写)/:protected(默认,属主写、大家读)/:private(只有属主)。缓存场景的标配就是示例里那行:set + named_table + public + read_concurrency。

要点:①读写不经过任何进程的邮箱,任意进程直接访问,配合 read_concurrency 多核并行读,吞吐是 GenServer 的几个数量级之上;②单条操作原子,还有 update_counter 这类原子计数器(限流器的地基);③表有属主进程,属主死,表陪葬——所以生产上表要由监督树里的稳定进程持有。

除了按键查,ETS 还支持按模式查——把第 3 章的直觉搬进表里::ets.match_object(:vitals_cache, {:_, %{hr: :_}}) 捞出所有带心率字段的行;更复杂的筛选走 :ets.select + match spec(一种编译进虚拟机的查询谓词,第 12 章的追踪器底层用的同款机制)。另外两个近亲顺带备案::persistent_term——读零成本的全局只读项,适合「启动后几乎不变、读取极频繁」的配置(写入很贵,慎写);:counters——纯计数场景比 ETS 更快的原子计数器数组。

⚠ 病历警示

意识到没有?ETS 是可变的共享状态——我们花五章建立的不可变教义,在这里开了一扇官方后门。开门是因为真实需求(高频共享读),但请守规矩:①ETS 只放「查表型」数据(缓存、配置、计数、注册表),业务状态机还是住 GenServer;②跨多条 ETS 操作没有事务,「先 lookup 再 insert」存在竞态,要原子性就回 GenServer 串行化,或用 update_counter 这类原子原语。口诀:ETS 是仪表板,不是病历本。

Registry + DynamicSupervisor:每个订单一个进程

第 7 章吹过的牛(「每个用户一个进程」)现在落地。两个问题要解:进程是运行时按需出生的(监督树的 children 清单是静态的),以及出生后怎么按业务 ID 找到它。答案各一个标准件:

children = [
  {Registry, keys: :unique, name: Ward.Reg},          # 电话簿
  {DynamicSupervisor, name: Ward.OrderSup}            # 产房
]

# 新订单来了:动态拉起一个进程,名字登记为 {:order, id}
DynamicSupervisor.start_child(Ward.OrderSup,
  {Ward.Order, name: {:via, Registry, {Ward.Reg, {:order, id}}}})

# 任何地方按业务 ID 喊话,不用传 PID
GenServer.call({:via, Registry, {Ward.Reg, {:order, "A-1024"}}}, :status)

DynamicSupervisor 是产房版监督者:不预设孩子清单,运行时随到随生,每个孩子照常享受重启策略。Registry 是本地电话簿:业务键 → PID,进程死亡自动销号(基于 monitor),重启后新 PID 重新登记,呼叫方永远拨「号码」不拨「尸体」。这对组合是 Phoenix 频道、游戏房间、IoT 设备会话的通用底盘——每个业务实体一个进程、一份状态、一条串行队列,竞态在架构层面消失。

选型表:并发工具四件套

工具一句话人设典型场景别用它做
Task跑腿的,干完即散外包慢活、并发批量拉取长活服务、持状态
Agent伞架,只存不办简单共享值、配置有业务逻辑的状态
GenServer心脏,守着状态办业务状态机、串行化写、协调者高频只读查表、搬重物
ETS仪表板,人人可读缓存、计数、注册表需要事务的多步修改
⚕ 查房记录

压轴的架构观,来自 José Valim 本人反复强调的一句话:进程是并发单元,不是代码组织单元。新手最常见的误用,是给「用户模块」「订单模块」各配一个 GenServer 来「分层」——结果把并行的请求全都串行进几个单点,性能不如不用。判据永远是运行时的:这里有没有需要串行化保护的状态?有没有需要独立崩溃、独立重启的活动?都没有,就写普通模块和纯函数——第 4 章的那种。九成的代码应该是纯函数,进程只出现在并发的骨节上。

▸ Android 类比

对表一下:Task.async/await ≈ 协程的 async/await,连超时语义都同构;async_stream ≈ flatMapMerge(concurrency = n),但一行顶一套;ETS ≈ 进程内的 ConcurrentHashMap(无锁读的味道);Registry + DynamicSupervisor 的「按 ID 找会话」,像极了你用 Map<SessionId, Scope> 手工管理的协程作用域——只是这里的「作用域」死了会自动重启、自动销号。

✚ 动手自测
  1. 用 Task.async_stream 并发抓 20 个 URL(拿 httpbin 练手),把 max_concurrency 从 1 调到 20,感受耗时曲线。
  2. 用 ETS + update_counter 写一个每秒限 100 次的限流器,对比你第 8 章用 GenServer 写的版本,谁的读路径短?
  3. 设计题:聊天应用的「房间」——进程、状态、注册、监督各用什么件?画出那棵树。
  4. 复述查房记录那句话,并举一个「不该用进程」的反例。

工具箱最后压一句观测预告:这四件套全部是「玻璃做的」——Task 是进程(observer 里看得见)、Agent 是 GenServer(:sys.get_state 掏得出)、ETS 表在 :observer 的 Table Viewer 页签里整表可浏览、Registry 本身也是查询接口。你即将进入的 VOL.IV 不需要引入任何新的「探针」——探的就是这些你已经认识的东西。

心跳卷完结。你已经有了一套能自愈的并发系统——现在,是时候兑现全书的承诺了:下一章,连进一个正在跳动的节点,把手伸进去。