Go语言并行读取多文件遇死锁：如何正确关闭通道？

问题核心原因 #

你的代码死锁是因为外层循环for donecnt < filecnt嵌套了for s := range res：range会一直阻塞直到通道关闭，但你要等所有文件处理完才关闭通道，形成了循环等待——内层range等通道关闭，外层循环等donecnt达标，可donecnt的更新依赖内层range读取空字符串，而内层range一旦启动就不会退出，根本没机会去判断donecnt。

基础修复方案：用sync.WaitGroup管理goroutine #

通过sync.WaitGroup跟踪所有读文件的goroutine，等全部完成后再关闭通道，让range res能正常遍历后退出。

修改后的代码：

package main

import (
    "bufio"
    "os"
    "strings"
    "sync"
)

func main() {
    dirname := "."

    res := make(chan string)
    var wg sync.WaitGroup

    dir, err := os.Open(dirname)
    if err != nil {
        panic(err)
    }
    defer dir.Close()

    files, err := dir.ReadDir(10)
    if err != nil {
        panic(err)
    }

    for _, file := range files {
        // 用HasPrefix替代Index，逻辑更直观
        if strings.HasPrefix(file.Name(), "input_") {
            wg.Add(1)
            // 循环内启动goroutine时直接传参，避免循环变量引用问题
            go func(filename string) {
                defer wg.Done()
                readFile(filename, res)
            }(dirname + "/" + file.Name())
        }
    }

    // 单独开goroutine等待所有任务完成后关闭通道
    go func() {
        wg.Wait()
        close(res)
    }()

    // 遍历通道内容直到关闭
    for s := range res {
        if s != "" {
            // 这里可根据需求处理读取到的文本，比如打印、存储
            println(s)
        }
    }
}

func readFile(filename string, res chan string) {
    file, err := os.Open(filename)
    if err != nil {
        println("打开文件失败:", err)
        return
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        res <- scanner.Text()
    }

    // 处理扫描过程中的错误
    if err := scanner.Err(); err != nil {
        println("扫描文件失败:", err)
    }
    // 不再需要发送空字符串标记完成，WaitGroup已跟踪状态
}

优化点说明 #

• 替换手动计数为sync.WaitGroup：避免手动维护donecnt的逻辑错误，更可靠地跟踪goroutine生命周期。
• 单独goroutine关闭通道：确保所有读文件任务完成后才关闭通道，range能正常退出。
• 简化文件名判断逻辑：用strings.HasPrefix替代strings.Index，代码可读性更强。
• 添加错误处理：原代码忽略所有错误，实际项目中必须处理文件打开、读取的异常，避免隐藏问题。
• 修复goroutine参数陷阱：循环内启动goroutine时直接传递文件名，避免多个goroutine引用同一循环变量的问题。

进阶优化：限制并发数（适合大量文件场景） #

如果待读取的文件数量很多，无限制启动goroutine会占用过多系统资源，可通过信号量限制并发数：

package main

import (
    "bufio"
    "os"
    "strings"
    "sync"
)

func main() {
    dirname := "."
    maxConcurrency := 5 // 限制最大同时运行的goroutine数量

    res := make(chan string)
    // 用带缓冲的通道做信号量，控制并发
    sem := make(chan struct{}, maxConcurrency)
    var wg sync.WaitGroup

    dir, err := os.Open(dirname)
    if err != nil {
        panic(err)
    }
    defer dir.Close()

    // 读取目录下所有文件，而非仅10个
    files, err := dir.ReadDir(-1)
    if err != nil {
        panic(err)
    }

    for _, file := range files {
        if strings.HasPrefix(file.Name(), "input_") && !file.IsDir() {
            wg.Add(1)
            sem <- struct{}{} // 获取信号量，达到上限则阻塞
            go func(filename string) {
                defer func() {
                    wg.Done()
                    <-sem // 释放信号量
                }()
                readFile(filename, res)
            }(dirname + "/" + file.Name())
        }
    }

    go func() {
        wg.Wait()
        close(res)
    }()

    for s := range res {
        println(s)
    }
}

func readFile(filename string, res chan string) {
    file, err := os.Open(filename)
    if err != nil {
        println("打开文件失败:", err)
        return
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        res <- scanner.Text()
    }

    if err := scanner.Err(); err != nil {
        println("扫描文件失败:", err)
    }
}

这个版本通过信号量控制并发数，避免系统资源被过度占用，适合处理大量文件的场景。

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