如何为DynamicSupervisor实现长运行任务的子进程？

现有代码 #

MyWorker 模块 #

defmodule MyWorker do
  def do_long_running_work(a, b, c) do
    # ......
  end
end

MyDynamicSupervisor 模块 #

defmodule MyDynamicSupervisor do
  use DynamicSupervisor

  def start_link(_arg) do
    DynamicSupervisor.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    DynamicSupervisor.init(strategy: :one_for_one)
  end

  def add_my_worker(worker_name, game_id) do
    child_spec = {MyWorker, {worker_name, game_id}}
    DynamicSupervisor.start_child(__MODULE__, child_spec)
  end

  def remove_my_worker(worker_pid) do
    DynamicSupervisor.terminate_child(__MODULE__, worker_pid)
  end

  def children do
    DynamicSupervisor.which_children(__MODULE__)
  end

  def count_children do
    DynamicSupervisor.count_children(__MODULE__)
  end
end

问题与解答 #

1. 适用于长时任务的 MyWorker 实现方案 #

GenServer 基于单进程消息循环模型，长时任务会阻塞消息处理，完全不适合你的场景。推荐两种实用方案：

方案一：Task + Agent 封装（适合一次性无状态长任务）

如果任务启动后无需中途交互，直接运行到结束，可以用 Task 执行长任务，再用轻量的 Agent 持有任务引用，方便后续管理：

defmodule MyWorker do
  def start_link({worker_name, game_id}) do
    # 用 async_nolink 避免父进程被任务崩溃牵连
    task = Task.async_nolink(fn ->
      do_long_running_work(worker_name, game_id, :default_arg)
    end)
    
    # 启动Agent持有任务元数据，便于查询状态
    Agent.start_link(fn -> %{task: task, worker_name: worker_name, game_id: game_id} end)
  end

  def do_long_running_work(name, game_id, arg) do
    # 模拟数小时的长任务逻辑
    Process.sleep(:timer.hours(2))
    IO.puts("任务 #{name}/#{game_id} 执行完成")
  end
end

方案二：GenStage（适合流式/可控制的长任务）

如果任务需要中途暂停、处理数据流或做负载均衡，GenStage 是更优选择。它支持持续数据流处理，不会阻塞进程，天然适配长时间运行的场景。

2. 不使用 GenServer 时，MyWorker 的 start_link 实现 #

start_link 只需符合 OTP 进程启动规范（返回 {:ok, pid} 或 {:error, reason}），无需依赖 GenServer。以下是两种实现方式：

方式一：自定义裸进程

自己实现消息循环，适合简单场景：

defmodule MyWorker do
  def start_link(args) do
    pid = spawn_link(fn -> loop(args) end)
    {:ok, pid}
  end

  defp loop(args) do
    receive do
      {:get_status, sender} ->
        send(sender, {:status, :running, args})
        loop(args)
      {:stop, reason} ->
        exit(reason)
      _ ->
        loop(args)
    end
  end

  def do_long_running_work(a, b, c) do
    # 长任务逻辑
  end
end

方式二：直接启动长任务进程

让 start_link 直接启动长任务，并确保进程启动成功：

defmodule MyWorker do
  def start_link({worker_name, game_id}) do
    parent = self()
    pid = spawn_link(fn ->
      send(parent, {:worker_started, self()})
      do_long_running_work(worker_name, game_id, :arg)
    end)
    
    receive do
      {:worker_started, ^pid} -> {:ok, pid}
    after 5000 -> {:error, :timeout}
    end
  end

  def do_long_running_work(a, b, c) do
    # 长任务逻辑
  end
end

3. 数千实例场景下，DynamicSupervisor 的适用场景 #

DynamicSupervisor 完全匹配你的需求，它就是为动态创建、销毁大量子进程设计的：

• 进程管理：支持自动重启异常崩溃的子进程（通过 :one_for_one 等策略），也可主动调用 terminate_child 销毁进程；若任务崩溃后无需重启，可在子进程的 child_spec 中设置 restart: :temporary。
• 状态与统计：你已实现的 children/0 和 count_children/0 可直接使用，前者返回所有子进程信息，后者返回活跃进程数、重启次数等统计数据。
• 高并发支撑：Erlang VM 原生支持数千甚至数万个进程，DynamicSupervisor 作为监管进程的性能开销极低，不会成为瓶颈。

仅需注意：若每个长任务进程占用大量资源，需控制实例数量，避免耗尽系统资源。

内容的提问来源于stack exchange，提问作者Koklushkin