本文介绍如何通过 goroutine 并发模型与 bufio.Scanner 结合，优化大型 CSV 文件的 ETL 处理流程，包括提取、转换、加载三阶段的并行化设计及关键注意事项。
本文介绍如何通过 goroutine 并发模型与 `bufio.scanner` 结合，优化大型 csv 文件的 etl 处理流程，包括提取、转换、加载三阶段的并行化设计及关键注意事项。
在处理大型 CSV 文件（例如每行约 300 字节、总大小达 GB 级）时，串行读取—解析—入库极易成为性能瓶颈。Go 语言原生支持轻量级并发，合理使用 goroutine 可显著提升吞吐量。但需注意：并发 ≠ 并行，更不等于盲目增加 goroutine 数量。
推荐架构：ETL 三阶段流水线（Pipeline）
将 ETL 拆分为三个独立职责的 goroutine，通过有缓冲 channel 协作，形成高效流水线：
- Extract（E）：单 goroutine 顺序读取文件，逐行发送至 chan string（避免多 goroutine 竞争文件偏移）；
- Transform（T）：N 个 goroutine 并发消费原始行，执行 strings.Split()、字段校验、结构体映射、JSON 序列化等 CPU 密集操作；
- Load（L）：M 个 goroutine 异步写入数据库（如使用 sql.DB 的连接池），避免阻塞上游。
func runETLPipeline(filename string, workers int) error {
// 1. Extract: 单 goroutine 安全读取
lines := make(chan string, 1024)
go func() {
defer close(lines)
f, err := os.Open(filename)
if err != nil {
panic(err)
}
defer f.Close()
scanner := bufio.NewScanner(f)
for scanner.Scan() {
lines <h3>关于“从任意位置随机读取”的说明</h3><p>bufio.Scanner <strong>不支持随机偏移读取</strong>——它依赖底层 io.Reader 的顺序性，且内部维护缓冲区状态。若需分片并行读取（如多 goroutine 同时处理不同文件段），应：</p><div class="aritcle_card flexRow artxards">
<div class="artcardd flexRow">
<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill6686" title="QuickBooks Mileage CSV"><img
src="https://img.php.cn/upload/skill/000/000/081/179108649715165.jpg" alt="QuickBooks Mileage CSV" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a rel="nofollow" href="/xiazai/skill6686" title="QuickBooks Mileage CSV" class="overflowclass">QuickBooks Mileage CSV</a>
<p class="overflowclass">使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。</p>
</div>
<a rel="nofollow" href="/xiazai/skill6686" title="QuickBooks Mileage CSV" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
</a>
</div>
</div>
- 使用 os.Seek() 定位到字节偏移；
- 手动按行切分：跳过不完整行（确保从 \n 后开始），再用 bufio.NewReader(io.LimitReader(...)) 构建局部扫描器；
- 更推荐方案：预处理生成行偏移索引（如记录每行起始位置），或改用 encoding/csv.Reader 配合 io.Seeker（需自行处理换行边界）。
⚠️ 注意事项：
- 避免无缓冲 channel 导致 goroutine 死锁；
- scanner.Text() 返回的字符串指向内部缓冲区，若需长期持有，请 string(scanner.Bytes()) 拷贝；
- 数据库写入务必复用连接池，禁用 defer db.Close() 在 goroutine 内；
- 监控 goroutine 泄漏：使用 runtime.NumGoroutine() 或 pprof。
总结：合理划分 ETL 阶段、控制 worker 数量（建议从 runtime.NumCPU() 开始调优）、配合缓冲 channel 与资源复用，即可在低内存开销下实现高吞吐 ETL。并发的本质是解耦阻塞点，而非堆砌 goroutine。
