如何在Elixir中实现数据库查询分页并分批返回数据到前端

HTTP协议本身是单次请求单次响应的模式，无法在一个同步请求里多次返回内容给前端，根据常用技术栈有两种成熟实现路径：

方案1：WebSocket实时推送（推荐，用户体验最好） #

如果你的项目用了Phoenix Channel，优先选这个方案：

1. 把原递归逻辑拆成异步任务，每次处理完500条就通过Channel把当前批次的记录推送给前端
2. 所有批次处理完成后再推送结束信号给前端

修改后的代码示例：

# 首先调整处理函数，支持传入推送回调
def seen_page(page_id, type \\ "ALL", push_callback, total \\ 0) do
  repo = Shard.by_page(page_id)
  
  # 先查当前批次的记录，再做更新，避免直接update_all拿不到记录内容
  query = 
    if type === "ALL" do
      from(c in Conversation, where: c.page_id == ^page_id and c.seen == false, limit: 500)
    else
      from(c in Conversation, where: c.page_id == ^page_id and c.seen == false and c.type == ^type, limit: 500)
    end

  # 获取当前要更新的记录
  records = repo.all(query)
  
  if Enum.empty?(records) do
    # 处理完成，推送结束信号
    push_callback.({:finish, total})
    Cache.update_page_unread_count_limit_2k(page_id)
    IO.puts "Đã đánh dấu tất cả #{type} của trang #{page_id} thành đã đọc"
    total
  else
    # 批量更新这批记录
    record_ids = Enum.map(records, & &1.id)
    from(c in Conversation, where: c.id in ^record_ids)
    |> repo.update_all([set: [seen: true, unread_count: 0]])
    
    # 推送当前批次给前端
    push_callback.({:batch, records})
    
    # 递归处理下一批
    seen_page(page_id, type, push_callback, total + length(records))
  end
end

# Phoenix Channel 里的调用示例
def handle_in("start_mark_seen", %{"page_id" => page_id, "type" => type}, socket) do
  # 异步启动处理任务，避免阻塞Channel
  Task.start(fn ->
    push_callback = fn 
      {:batch, records} -> push(socket, "batch_processed", %{records: records})
      {:finish, total} -> push(socket, "all_done", %{total: total})
    end
    seen_page(page_id, type, push_callback)
  end)
  
  {:reply, :ok, socket}
end

方案2：HTTP分块流式返回 #

如果不想用WebSocket，用Phoenix的分块响应也可以实现单次请求逐批返回内容：

# Controller 中的代码示例
def mark_seen(conn, %{"page_id" => page_id, "type" => type}) do
  conn = send_chunked(conn, 200)
  
  total = process_batches(conn, page_id, type, 0)
  
  # 最后返回总数量
  chunk(conn, Jason.encode!(%{type: "finish", total: total}) <> "\n")
  
  conn
end

defp process_batches(conn, page_id, type, total) do
  repo = Shard.by_page(page_id)
  query = 
    if type === "ALL" do
      from(c in Conversation, where: c.page_id == ^page_id and c.seen == false, limit: 500)
    else
      from(c in Conversation, where: c.page_id == ^page_id and c.seen == false and c.type == ^type, limit: 500)
    end
  
  records = repo.all(query)
  
  if Enum.empty?(records) do
    Cache.update_page_unread_count_limit_2k(page_id)
    total
  else
    record_ids = Enum.map(records, & &1.id)
    from(c in Conversation, where: c.id in ^record_ids)
    |> repo.update_all([set: [seen: true, unread_count: 0]])
    
    # 把当前批次返回给前端
    chunk(conn, Jason.encode!(%{type: "batch", records: records}) <> "\n")
    
    process_batches(conn, page_id, type, total + length(records))
  end
end

注意事项 #

• 原来的代码直接用update_all无法拿到更新的记录内容，所以必须先查再更新，才能拿到要返回给前端的记录数据
• 如果数据量特别大，可以给Conversation表的page_id、seen、type字段加联合索引，大幅提升查询速度
• 异步处理的时候要注意数据库事务一致性，如果担心更新失败可以给每批处理加事务包裹

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