React+Golang GraphQL服务全量数据下载遇connection reset by peer解决方案

一、可行解决选项 #

1. 前端分页聚合导出 #

• 逻辑：前端复用现有分页查询逻辑，循环拉取全量数据到内存，聚合后生成CSV下载。
• 优缺点：无需修改后端，但数据量极大时会导致前端内存溢出，且多请求易触发限流，用户等待时间长。

2. 后端GraphQL流式响应（推荐复用方案） #

• 逻辑：在GraphQL体系内实现流式输出，复用现有分页查询Resolver，后端边查边生成CSV流返回。
• 优缺点：完全复用GraphQL的查询、过滤逻辑，保持接口通用性；需调整后端GraphQL处理逻辑，前端需适配流式响应。

3. 专用导出端点复用GraphQL Resolver #

• 逻辑：新增独立导出端点，但内部直接调用现有GraphQL的分页查询Resolver，通过Gin流式输出CSV。
• 优缺点：前端实现简单，无需改动Relay；需维护额外端点，但业务逻辑完全复用，无重复代码。

4. 异步导出+通知 #

• 逻辑：后端接收导出请求后异步处理，生成CSV文件存储，完成后通知用户下载。
• 优缺点：适合超大数据量，不占用用户等待时间；需额外实现存储、通知机制，复杂度较高。

二、最优复用代码方案（两种实现路径） #

路径1：GraphQL流式响应（保持GraphQL通用性） #

后端实现步骤

1. 扩展GraphQL Schema：添加CSV流类型的导出字段，复用现有过滤输入类型

type Query {
  exportDataAsCSV(filter: DataFilterInput!): CSVStream
}

scalar CSVStream

2. 实现流式Resolver：复用现有分页查询逻辑，循环拉取数据并写入响应流

func (r *queryResolver) ExportDataAsCSV(ctx context.Context, filter *DataFilterInput) (interface{}, error) {
  w := ctx.Value("response_writer").(io.Writer)
  page := 1
  pageSize := 1000
  firstWrite := true
  csvWriter := csv.NewWriter(w)

  // 写入表头
  if firstWrite {
    if err := csvWriter.Write([]string{"ID", "Name", "CreatedAt"}); err != nil {
      return nil, err
    }
    csvWriter.Flush()
    firstWrite = false
  }

  // 循环分页拉取数据
  for {
    // 复用现有分页查询Resolver
    conn, err := r.ListData(ctx, filter, &PageInput{Page: page, Size: pageSize})
    if err != nil {
      return nil, err
    }
    if len(conn.Nodes) == 0 {
      break
    }

    // 转换数据为CSV行并写入流
    for _, item := range conn.Nodes {
      row := []string{
        strconv.Itoa(item.ID),
        item.Name,
        item.CreatedAt.Format(time.RFC3339),
      }
      if err := csvWriter.Write(row); err != nil {
        return nil, err
      }
    }
    csvWriter.Flush()
    page++
    time.Sleep(50 * time.Millisecond) // 控制数据库查询速率
  }
  return nil, nil
}

3. 修改Gin GraphQL端点：识别导出查询，切换为流式响应模式

func GraphQLHandler(c *gin.Context) {
  var req struct {
    Query     string                 `json:"query"`
    Variables map[string]interface{} `json:"variables"`
  }
  if err := c.ShouldBindJSON(&req); err != nil {
    c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
    return
  }

  // 判断是否为导出查询
  if strings.Contains(req.Query, "exportDataAsCSV") {
    c.Header("Content-Type", "text/csv")
    c.Header("Content-Disposition", "attachment; filename=data.csv")
    
    // 将响应Writer传入GraphQL上下文
    ctx := context.WithValue(c.Request.Context(), "response_writer", c.Writer)
    result := graphql.Do(graphql.Params{
      Schema:        schema,
      RequestString: req.Query,
      Variables:     req.Variables,
      Context:       ctx,
    })
    
    if len(result.Errors) > 0 {
      c.Writer.Write([]byte("Export error: " + result.Errors[0].Message))
    }
    return
  }

  // 常规GraphQL响应逻辑
  result := graphql.Do(graphql.Params{
    Schema:        schema,
    RequestString: req.Query,
    Variables:     req.Variables,
    Context:       c.Request.Context(),
  })
  c.JSON(http.StatusOK, result)
}

前端实现：用Fetch处理流式响应（Relay默认不支持流式，需单独实现）

async function exportViaGraphQL(filter) {
  const query = `
    query ExportData($filter: DataFilterInput!) {
      exportDataAsCSV(filter: $filter)
    }
  `;

  const response = await fetch('/data', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ query, variables: { filter } }),
  });

  if (!response.ok) throw new Error('Export failed');

  const reader = response.body.getReader();
  const blobParts = [];
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    blobParts.push(value);
  }

  const blob = new Blob(blobParts, { type: 'text/csv' });
  const url = URL.createObjectURL(blob);
  const a = document.createElement('a');
  a.href = url;
  a.download = 'data.csv';
  a.click();
  URL.revokeObjectURL(url);
}

路径2：专用导出端点复用Resolver（快速实现） #

后端实现：新增导出端点，直接调用现有Resolver

func ExportCSVHandler(c *gin.Context) {
  var req struct {
    Filter DataFilterInput `json:"filter"`
  }
  if err := c.ShouldBindJSON(&req); err != nil {
    c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
    return
  }

  // 设置响应头
  c.Header("Content-Type", "text/csv")
  c.Header("Content-Disposition", "attachment; filename=data.csv")

  // 流式输出CSV
  err := c.Stream(func(w io.Writer) bool {
    page := 1
    pageSize := 1000
    firstWrite := true
    csvWriter := csv.NewWriter(w)

    // 写入表头
    if firstWrite {
      csvWriter.Write([]string{"ID", "Name", "CreatedAt"})
      csvWriter.Flush()
      firstWrite = false
    }

    // 复用现有分页Resolver
    for {
      conn, err := r.ListData(c.Request.Context(), &req.Filter, &PageInput{Page: page, Size: pageSize})
      if err != nil {
        w.Write([]byte("Error: " + err.Error()))
        return false
      }
      if len(conn.Nodes) == 0 {
        break
      }

      // 写入数据行
      for _, item := range conn.Nodes {
        csvWriter.Write([]string{
          strconv.Itoa(item.ID),
          item.Name,
          item.CreatedAt.Format(time.RFC3339),
        })
      }
      csvWriter.Flush()
      page++
      time.Sleep(50 * time.Millisecond)
    }
    return false
  })

  if err != nil {
    c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
  }
}

前端实现：调用专用端点

async function exportCSV(filter) {
  const response = await fetch('/export-csv', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ filter }),
  });

  if (!response.ok) throw new Error('Export failed');

  const blob = await response.blob();
  const url = URL.createObjectURL(blob);
  const a = document.createElement('a');
  a.href = url;
  a.download = 'data.csv';
  a.click();
  URL.revokeObjectURL(url);
}

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