本文介绍如何使用 go 的 goroutine、channel 和 waitgroup 协同工作，安全高效地并发处理 5 个各含 400 万行的日志文件，避免竞态、资源泄漏与阻塞，兼顾吞吐与内存可控性。
本文介绍如何使用 go 的 goroutine、channel 和 waitgroup 协同工作，安全高效地并发处理 5 个各含 400 万行的日志文件，避免竞态、资源泄漏与阻塞，兼顾吞吐与内存可控性。
在 Go 中并发处理多个超大日志文件（如每个 400 万行），关键不在于“越多 goroutine 越好”，而在于职责分离、流式控制与资源收敛。原始方案中嵌套 wg1/wg2、在循环内无节制启动 goroutine、以及对 defer wg2.Done() 的误用，会导致严重问题：fScanner.Text() 在闭包中被多 goroutine 共享引用，引发数据竞争；wg2.Wait() 阻塞整个文件处理流程，彻底丧失并发性；且未关闭 channel 或管理输出缓冲，易造成 goroutine 泄漏。
正确的做法是采用 “生产者–多消费者–单写入”管道模型：
- 每个文件由独立 goroutine 处理（processFile）；
- 行解析与业务逻辑作为生产者，通过带缓冲 channel（如 chan string, 10）异步投递结果；
- 多个专用写入 goroutine（如 numGoWriters = 5~10）从 channel 消费并批量写入目标文件，利用 bufio.Writer 提升 I/O 效率；
- 使用单一 sync.WaitGroup 精确等待所有行处理完成，再关闭 channel，自然终止写入协程。
以下是可直接运行的核心实现（已修复原始代码中的关键缺陷）：
func processRow(r string, ch chan<p>main 函数应为每个输入文件启动独立 processFile，并用 WaitGroup 统一协调：</p><pre class="brush:php;toolbar:false;">func main() {
filenames := []string{"log1.txt", "log2.txt", "log3.txt", "log4.txt", "log5.txt"}
var wg sync.WaitGroup
for i, fname := range filenames {
inFile, err := os.Open(fname)
if err != nil {
log.Printf("skip %s: %v", fname, err)
continue
}
// 注意：此处不能 defer inFile.Close() —— 它会在 main 返回时才执行，而 processFile 需要实时读取
// 正确做法是在 processFile 内部打开并关闭输入文件，或传入路径而非 *os.File
wg.Add(1)
go func(name string, idx int) {
defer wg.Done()
outName := fmt.Sprintf("processed_%d.txt", idx+1)
processFileByPath(name, outName) // 推荐：processFile 接收文件路径，在内部打开/关闭
}(fname, i)
}
wg.Wait()
fmt.Println("All files processed.")
}关键注意事项：
✅ 始终拷贝循环变量：line := scanner.Text(); go func(l string){...}(line)，否则闭包共享同一变量地址；
✅ 合理设置 channel 缓冲：过小（如 1）导致生产者频繁阻塞，过大（如 100000）浪费内存；建议从 50~500 开始压测调整；
✅ 输入文件应在 processFile 内打开/关闭：避免 main 中 defer inFile.Close() 错误延迟关闭；
✅ 写入协程必须 defer w.Flush()：防止最后一部分缓冲未写出；
✅ 添加错误处理与日志：尤其在 I/O 和 scanner 扫描阶段，不可静默失败。
该模式可轻松扩展至数十文件或更大数据量——只需调整写入协程数与 channel 缓冲，无需修改核心结构。它体现了 Go 并发哲学的精髓：用通信共享内存，以管道解耦关注点，靠同步原语确保确定性。
