Elixir 并发与 OTP

BEAM 进程 + GenServer + Supervisor——Elixir 区别于其他 FP 语言的独门。

这节课覆盖：

1. BEAM 进程是什么、跟 OS 进程和 goroutine 的区别
2. 消息传递（send / receive）
3. GenServer 状态机模式
4. Supervisor 监督树
5. Application 启动回调
6. Oban Worker 怎么用 GenServer 跑后台任务

**预计时间：**60 分钟。

一、BEAM 进程到底是什么？

BEAM 进程 = BEAM 调度器管理的极轻量级协程，跟 OS 进程 / goroutine / Java Thread 都不一样：

单位	初始内存	创建开销	隔离	可创建数（单核）
OS 进程（Linux）	~ MB	~ 100 μs	完全（独立地址空间）	~ 1 万
Java Thread	~ 512 KB	~ 10 μs	弱（共享堆）	~ 几千
Go goroutine	~ 2 KB	~ 1 μs	中（channel 同步）	~ 百万
BEAM 进程	~ 0.5 KB	~ 1 μs	完全（独立堆 + GC）	~ 百万+

**实际体验：**你可以创建 100 万个 BEAM 进程，只占用几 GB 内存。这就是 WhatsApp / Discord 选 Erlang 的原因。

二、最底层 API：spawn / send / receive

Elixir 暴露三个核心函数：

# 创建进程
pid = spawn(fn -> 1 + 1 end)
# pid 是 Erlang 进程 ID，类似 #PID<0.123.0>

# 进程间发消息（任何值都行）
send(pid, {:hello, "world"})

# 收消息（带模式匹配）
receive do
  {:hello, msg} -> "got: #{msg}"
  :stop -> :ok
after
  5000 -> "timeout"   # 5 秒没消息
end

这就是 Erlang 1986 年的设计：一切都是进程，进程之间只能消息传递。

类比 Go：go func() { ... }() ≈ spawn(fn -> ... end)，ch <- msg ≈ send(pid, msg)，msg := <-ch ≈ receive do ... end。

**类比 Akka (Scala/Java)：**几乎完全一样——Actor 模型就是 Erlang 模型的 Scala/Java 复刻。

三、消息 = 不可变值（写时复制）

消息发送时是值复制。但小对象（map、list、tuple）复制极快（几十纳秒），大 binary 走 **COW（Copy-On-Write）**共享：

big = String.duplicate("x", 10_000_000)  # 10 MB binary
send(pid, big)                            # 不复制 10 MB，共享同一份
# 当任何一方修改时，才真正复制

**类比：**Linux fork() 的 COW 内存页。BEAM 把它做在消息层。

四、GenServer：最常用的进程模式

spawn 裸用太原始——没有状态、没有标准化 API、没有生命周期。实际项目几乎都用 GenServer（Generic Server）：

defmodule Counter do
  use GenServer

  # 1. 启动（init 返回 {:ok, initial_state}）
  def start_link(arg) do
    GenServer.start_link(__MODULE__, arg, name: __MODULE__)
  end

  def init(_arg) do
    {:ok, 0}   # 初始 state
  end

  # 2. 同步 API（call）
  def get, do: GenServer.call(__MODULE__, :get)
  def increment, do: GenServer.call(__MODULE__, :inc)

  # 3. 异步 API（cast）
  def async_increment, do: GenServer.cast(__MODULE__, :inc)

  # 4. 回调（处理 call/cast）
  def handle_call(:get, _from, state) do
    {:reply, state, state}        # reply + 新 state
  end

  def handle_call(:inc, _from, state) do
    {:reply, state, state + 1}
  end

  def handle_cast(:inc, state) do
    {:noreply, state + 1}
  end
end

# 用法
{:ok, pid} = Counter.start_link([])
Counter.get           # 0
Counter.increment     # 1 (同步)
Counter.get           # 1
Counter.async_increment  # 异步，立即返回
Process.sleep(10)
Counter.get           # 2

项目例子：ex/elixir/lib/pipeline/jobs/news_digest.ex 里的 Oban Worker 就是一个 GenServer（use Oban.Worker 内部调用了 use GenServer）：

defmodule Pipeline.Jobs.NewsDigest do
  use Oban.Worker, queue: :digest, max_attempts: 3
  # ↑ use Oban.Worker 内部注入 GenServer 行为

  def perform(%Oban.Job{args: args}), do: run(args)  # ← 这是 GenServer 的 handle_call 等价的"任务入口"
end

核心要点：

1. 每个 GenServer 进程有 自己的 state（任何 Elixir term）
2. call（同步）— 等回复；cast（异步）— 不等
3. 回调 handle_call/handle_cast 是纯函数：接收 message + state，返回 {:reply, value, new_state} 或 {:noreply, new_state}
4. 所有 state 变更都在回调里发生——这意味着 没有共享内存、没有锁

五、link 和 trap_exit：进程间连接

当两个进程 link 之后，一个崩了，另一个会收到 EXIT 信号：

# GenServer 内部默认已经 link 到 caller
{:ok, pid} = GenServer.start_link(SomeMod, [])

# 如果 SomeMod 内部 crash：
# - 默认：caller 进程也死
# - 解决：trap_exit（GenServer 默认 trap）

# GenServer 是"被监督的"——它的崩溃由 Supervisor 负责
# GenServer 不该自己处理自己的崩溃

这就是 “let it crash” 哲学：进程不防御性 if，崩了让 supervisor 重启。这跟 Java 满屏 try-catch 是相反的思路。

六、Supervisor：监督树

一个 Supervisor 管理一组子进程，负责在它们崩溃时重启：

defmodule MyApp.Supervisor do
  use Supervisor

  def start_link(arg) do
    Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
  end

  def init(_arg) do
    children = [
      {Counter, []},                          # child spec: 模块 + 启动参数
      {Pipeline.Repo, []},                    # Ecto Repo
      {Pipeline.Jobs.NewsDigest, []},         # Oban Worker（虽然实际由 Oban 启动）
    ]

    opts = [strategy: :one_for_one]   # 一个崩了不影响其他
    Supervisor.init(children, opts)
  end
end

4 种重启策略：

策略	行为
:one_for_one	一个崩了，只重启那个（默认）
:one_for_all	一个崩了，全部重启
:rest_for_one	一个崩了，它和它之后启动的全部重启
:temporary	不重启（child spec 设置）

项目例子（lib/pipeline/application.ex）：

defmodule Pipeline.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [Pipeline.Repo, {Finch, name: Pipeline.Finch}]

    children =
      if Application.get_env(:pipeline, :start_oban, true) do
        children ++ [{Oban, Application.fetch_env!(:pipeline, Oban)}]
      else
        children
      end

    children =
      if Application.get_env(:pipeline, :start_web, true) do
        port = Application.get_env(:pipeline, :web_port, 8002)
        ip = Application.get_env(:pipeline, :web_ip, {127, 0, 0, 1})
        children ++ [{Bandit, plug: Pipeline.Web, port: port, ip: ip}]
      else
        children
      end

    opts = [strategy: :one_for_one, name: Pipeline.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

这就是整个应用的”启动图”：

• Pipeline.Repo（Ecto DB 池）
• Finch（HTTP 客户端，命名 Pipeline.Finch）
• Oban（后台任务队列，配置从 config/config.exs 读）
• Bandit（HTTP 服务器，绑 127.0.0.1:8002，plug 是 Pipeline.Web）

任何一个崩了，:one_for_one 只重启那个，其他不受影响。

七、Application 回调

mix.exs 指定 mod: {Pipeline.Application, []} 告诉 VM：“启动时调用 Pipeline.Application.start/2”。

Application 模块的 start/2 必须返回 {:ok, pid}——这个 pid 就是 整个监督树的根 supervisor。

八、Oban Worker 完整生命周期

Oban 是个 基于 PG 的持久化任务队列（任务存在 oban_jobs 表）。use Oban.Worker 等价于 use GenServer 加上”任务到达时调用 perform/1”。

完整例子（lib/pipeline/jobs/news_digest.ex）：

defmodule Pipeline.Jobs.NewsDigest do
  use Oban.Worker, queue: :digest, max_attempts: 3
  require Logger

  alias Pipeline.NewsDigest.Context

  # 入口（Oban 调度时调用）
  def perform(%Oban.Job{args: args}), do: run(args)
  def perform(args) when is_map(args), do: run(args)

  # 实际业务逻辑
  def run(args) when is_map(args) do
    period = Map.get(args, "period") || Map.get(args, :period, "close")
    role = Map.get(args, "role") || Map.get(args, :role, period_to_role(period))
    is_closing = Map.get(args, "is_closing", false) || Map.get(args, :is_closing, false)

    {:ok, run_id} = Pipeline.Execution.start("news_digest")

    try do
      text = Context.window_text(period)

      case Pipeline.Llm.summarize_digest(text) do
        {:ok, summary} ->
          row_count = if is_closing, do: Context.upsert_today_digest(summary), else: 0

          push_result =
            if is_closing do
              Context.maybe_push(summary, role)
            else
              :ok
            end

          if Context.needs_research?(summary) do
            maybe_enqueue_research(summary)
          end

          case push_result do
            :ok ->
              Context.mark_today_digest_as_pushed()
              Pipeline.Execution.finish(run_id, "done", row_count)
              :ok

            {:error, reason} ->
              Logger.warning("Wecom push failed: #{inspect(reason)} (digest still saved)")
              Pipeline.Execution.finish(run_id, "done", row_count, "push_failed: #{inspect(reason)}")
              :ok
          end

        {:error, reason} ->
          Pipeline.Execution.finish(run_id, "failed", 0, inspect(reason))
          {:error, reason}
      end
    rescue
      e ->
        Pipeline.Execution.finish(run_id, "failed", 0, Exception.message(e))
        {:error, e}
    end
  end

  defp maybe_enqueue_research(_summary), do: :ok

  defp period_to_role("pre_0715"), do: "pre"
  defp period_to_role("pre_0915"), do: "pre"
  defp period_to_role("trading_am"), do: "trading"
  defp period_to_role("trading_pm"), do: "trading"
  defp period_to_role("close"), do: "close"
  defp period_to_role(_), do: "资讯"
end

看懂这段代码你需要的概念都齐了：

• use Oban.Worker, queue: :digest, max_attempts: 3 — 注册为 Oban 任务，最多重试 3 次
• def perform(%Oban.Job{args: args}) — 任务入口（Oban 调用）
• Map.get(args, "period") || Map.get(args, :period, "close") — 兼容 string key 和 atom key
• try ... rescue e -> ... end — 异常兜底
• 所有结果都写 pipeline_execution 表（审计）

九、消息传递的”陷阱”和监控

死信：当 send 到一个不存在的 pid——消息直接丢失（BEAM 不报错）。这跟 Akka 的 deadLetter 一样。

**邮箱堆积：**如果一个进程收消息太慢，发件人一直 send，邮箱堆积。检测：

Process.info(pid, :message_queue_len)
# => {:message_queue_len, 12345}  # 危险信号！

项目里 Oban 用 PG 持久化消息，进程挂了消息不丢——这是 Oban 比裸 send 强的地方。

十、与其他并发模型速查

概念	Elixir/BEAM	Go	Java/Kotlin	Scala (Akka)
并发单位	BEAM 进程	goroutine	Thread	Actor
隔离	完全（独立堆）	共享堆	共享堆	独立（Akka）
消息	值复制（COW）	channel 引用	共享可变 + 锁	值复制
调度	抢占式（BEAM 调度器）	协作式（M:N）	OS 抢占	JDK ForkJoinPool
崩溃恢复	Supervisor 重启	手动	手动	Akka Supervisor
创建数	百万+	百万	几千	取决于 dispatcher

十一、测验（10 道选择 + 2 道读代码）

读代码题

**下一步：**看 Lesson 0003 · Elixir 独门机制（宏/协议/行为/热代码）。