本文介绍如何使用 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<- string) {
// 示例：简单转换为大写（替换为你的实际处理逻辑）
result := strings.ToUpper(r)
ch <- result
}
func writeRow(outFile *os.File, ch <-chan string) {
w := bufio.NewWriter(outFile)
defer w.Flush() // 确保最后缓冲区写出
for line := range ch {
if _, err := w.WriteString(line + "\n"); err != nil {
log.Printf("write error: %v", err)
return
}
}
}
func processFile(inFile *os.File, outPath string) {
outFile, err := os.Create(outPath)
if err != nil {
log.Printf("failed to create output file %s: %v", outPath, err)
return
}
defer outFile.Close()
ch := make(chan string, 100) // 缓冲区大小需根据内存与吞吐权衡
defer close(ch)
scanner := bufio.NewScanner(inFile)
var wg sync.WaitGroup
// 启动固定数量的写入协程
const numWriters = 8
for i := 0; i < numWriters; i++ {
go writeRow(outFile, ch)
}
// 启动行处理协程（每行一个，但受 channel 缓冲限制实际并发数）
for scanner.Scan() {
wg.Add(1)
line := scanner.Text() // 必须拷贝！避免闭包捕获迭代变量
go func(l string) {
defer wg.Done()
processRow(l, ch)
}(line)
}
if err := scanner.Err(); err != nil {
log.Printf("scan error: %v", err)
}
wg.Wait() // 等待所有行处理完毕
// channel 关闭后，writeRow 自然退出
}main 函数应为每个输入文件启动独立 processFile，并用 WaitGroup 统一协调：
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 并发哲学的精髓：用通信共享内存，以管道解耦关注点，靠同步原语确保确定性。
