Go 批量 CSV 导入怎么控内存：流式读取、资源预算和失败行回传实战

运营同学导入一份 18 万行的商品 CSV 后，接口没有立刻报错，进程的内存却一路往上走，最后被容器重启。问题通常不在 CSV 格式本身，而在服务端先把整个文件读完、再一次性组装全部记录。导入接口更像一条受控的传送带：边读、边校验、边落库，并把用户真正能修正的错误带回去。

先把“能上传”拆成三条资源预算

很多接口只在反向代理里配了 client_max_body_size，这只能挡住特别大的文件。一个 20MB 的 CSV 仍可能有几十万行；如果每行又带长备注，解析后的对象数量远比文件体积更值得担心。这里别急着提高内存，先把入口预算写清楚。

文件大小可以借助 http.MaxBytesReader 限住。它解决的是 HTTP 请求体，不负责 CSV 逻辑；行数和批次仍要在业务循环里单独计数。

const (
	maxUploadBytes = 32 

用 csv.Reader 把校验和写库放进同一条流水线

导入时不要先把所有行放进 []Product，再统一检查。正确的节奏是：读到一行就解析；字段不合法，记录一个用户可理解的提示；字段通过，放进当前批次；批次满了就写库并清空切片。这样堆内存主要受批次大小影响，而不是受文件总行数影响。

type RowProblem struct {
	Line   int    `json:"line"`
	Column string `json:"column"`
	Reason string `json:"reason"`
}

func readProducts(src io.Reader, save func([]Product) error) ([]RowProblem, error) {
	r := csv.NewReader(bufio.NewReaderSize(src, 64*1024))
	r.FieldsPerRecord = 4

	var batch []Product
	problems := make([]RowProblem, 0, 100)
	for line := 2; ; line++ { // 第一行是标题
		record, err := r.Read()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			return problems, fmt.Errorf("读取第 %d 行: %w", line, err)
		}
		if line-1 > maxRows {
			return problems, fmt.Errorf("数据行超过 %d 行", maxRows)
		}

		product, problem := parseProduct(record, line)
		if problem != nil {
			if len(problems)  0 {
		if err := save(batch); err != nil {
			return problems, err
		}
	}
	return problems, nil
}

这里有一个容易漏掉的边界：cap(problems) 是返回样本上限，不是停止校验的理由。若业务要求“有一行错就整份不入库”，就要把已通过的记录写到临时表或任务文件，确认没有错误后再做最终合并；若允许部分成功，响应里必须明确成功数、失败数和失败样本，别让用户以为整份都已生效。

失败行信息要让用户能直接改表格

“参数错误”对用户没有用。导入页最需要的是行号、列名和原因，例如“第 37 行：price 不能小于 0”。列名尽量使用 CSV 标题或业务中文名，不要把数据库字段 sale_price 原样丢给运营人员。

建议把接口结果设计成两层：同步响应只返回汇总和不超过 100 条样本；如果错误较多，生成一份带“错误原因”列的 CSV，供用户下载后修正。这样前端不必渲染几万条问题，服务端也不会因为错误响应太大而再次触碰预算。

type ImportResult struct {
	Accepted int          `json:"accepted"`
	Rejected int          `json:"rejected"`
	Problems []RowProblem `json:"problems"`
	ReportID string       `json:"report_id,omitempty"`
}

// 唯一索引仍由数据库保证；批量插入遇到冲突时，
// 把可定位的商品编码整理为行级问题，而不是吞掉错误。

写库前后各做一次核对，别只盯着接口状态码

流式读取并不等于导入一定可靠。第一层核对在应用内：记录读取行数、通过行数、失败行数和实际提交批次数；第二层核对在数据库侧：按本次 import_job_id 统计插入记录数，确认它等于接口返回的 Accepted。两边对不上时，先保留任务号和批次日志，再决定是否重试。

批量插入常见的两个坑也值得提前处理：

• 把每一行都开一个事务，失败定位很清楚，但吞吐会明显下降；通常按 200 到 1000 条做批次更平衡。
• 只在应用层检查商品编码重复。并发导入时仍可能撞车，唯一索引和冲突处理不能省。

若导入耗时超过 Web 请求的超时阈值，可以把上传文件落到受控存储，创建 import_job 后由后台任务处理。任务状态至少区分 queued、running、completed、failed，页面轮询时展示已处理行数。小文件同步处理更简单；把所有导入都强行异步，只会增加排障成本。

相关问题

CSV 的一行字段数量不一致时该跳过还是终止？

如果缺列会让业务含义无法判断，建议把该行列为失败并继续读取；如果 CSV 引号不闭合导致后续内容都不可信，则应终止本次导入并提示用户重新导出文件。

批量写入失败后需要回滚前面的批次吗？

取决于业务语义。库存初始化、价格模板这类要求全量一致的场景，应使用导入任务加临时表；允许部分成功的场景，则保留成功批次并准确返回失败原因。

为什么不直接把所有错误行都返回给前端？

错误可能和数据行一样多，完整回传会拖慢接口和页面。返回样本加下载报告，用户仍能定位问题，系统的内存和响应大小也更稳定。

导入接口需要做幂等控制吗？

需要。可以为每次上传生成导入任务号，并对文件摘要、业务日期或调用方请求号做重复判断；最终仍由唯一索引守住数据层边界。

把上限写在代码里，导入才有可预期的结果

一份 CSV 能否被安全处理，不取决于机器有多少内存，而取决于接口是否明确限制了文件、行数、批次和错误返回。逐行读取把峰值压到可控范围，批次写库让数据库压力有节奏，行级反馈则让用户能把表格改对。上线前拿一份接近上限的真实样本跑一遍，核对读取数、写入数和错误报告，往往比只压测一个成功接口更能发现问题。