Elixir中并行获取分页外部API并聚合数据的正确实现方式

嘿，你的思路方向是对的，但其实Agent并不是必须的——Elixir里有更简洁、更符合语言设计哲学的方式来处理这种并行任务并汇总结果，就是用Task.async_stream/3。让我给你拆解一下最佳实践：

核心方案：使用Task.async_stream #

Task.async_stream是Elixir处理批量异步任务的黄金工具——它会自动帮你管理并发数、处理任务超时，还能灵活处理结果返回逻辑。下面是具体实现步骤：

1. 定义单页数据获取函数 #

首先把单页API请求封装成一个独立函数，这里以常用的Tesla HTTP客户端为例：

defmodule APIFetcher do
  use Tesla

  # 替换成你的API基础地址
  plug Tesla.Middleware.BaseUrl, "https://api.example.com/your-endpoint"
  plug Tesla.Middleware.JSON

  def fetch_page(page) do
    case get("/", query: [page: page]) do
      {:ok, %Tesla.Env{status: 200, body: data}} -> {:ok, data}
      {:ok, %Tesla.Env{status: status}} -> {:error, "HTTP error: #{status}"}
      {:error, reason} -> {:error, "Request failed: #{inspect(reason)}"}
    end
  end
end

如果用HTTPoison也可以，核心是让函数返回:ok或:error元组，方便后续结果处理。

2. 并行执行并汇总结果 #

接下来生成需要请求的页码列表，用Task.async_stream并行处理，最后收集成功返回的数据：

defmodule DataAggregator do
  alias APIFetcher

  def fetch_all_pages(start_page, end_page) do
    pages = start_page..end_page

    pages
    |> Task.async_stream(
      &APIFetcher.fetch_page/1,
      max_concurrency: 10,  # 控制并发数，避免压垮API
      timeout: 10_000       # 设置超时时间，单位毫秒
    )
    |> Enum.reduce([], fn
      # 处理成功的任务结果，把数据加入汇总列表
      {:ok, {:ok, data}}, acc -> acc ++ data
      # 处理API请求失败的情况，打印日志并跳过
      {:ok, {:error, reason}}, acc ->
        IO.puts("Failed to fetch page: #{inspect(reason)}")
        acc
      # 处理任务本身失败（比如超时）的情况
      {:error, reason}, acc ->
        IO.puts("Task crashed: #{inspect(reason)}")
        acc
    end)
  end
end

调用方式非常简单：

# 获取1到10页的所有数据并汇总
all_combined_data = DataAggregator.fetch_all_pages(1, 10)

为什么不用Agent？ #

如果硬要用Agent实现当然可行，但会多一层不必要的状态管理——你需要手动启动Agent、让每个Task发送结果到Agent、最后再从Agent取出数据，不仅代码繁琐，还完全没必要（因为我们不需要共享可变状态，只是收集独立任务的结果）。

当然，如果你的需求是实时跟踪请求进度（比如展示已完成的请求数），Agent可以用来维护进度状态，但单纯的结果汇总场景，Task.async_stream是最优解。

关键注意事项 #

• 并发数控制：不要把max_concurrency设得过高，建议参考API文档的速率限制调整，避免被限流。
• 超时设置：根据API的响应速度设置合理的timeout，避免任务长时间挂起占用资源。
• 分页边界处理：如果不知道总页数，可以先请求第一页获取总页数，再生成对应的页码列表。
• 错误兜底：一定要处理各种失败情况，避免单个请求失败导致整个汇总流程崩溃。

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