多Worker模式Go CSV过滤程序运行死锁问题排查求助

问题概述 #

运行多Worker并行处理CSV行的Go程序时，触发死锁错误：

fatal error: all goroutines are asleep - deadlock!

错误发生在g.waitGroup.Wait()调用处，已尝试关闭输入通道但问题未解决。

根本原因分析 #

1. WaitGroup计数严重错误 #

• 初始化workerGroup时调用wg.Add(1)，后续每个worker重复执行wg.Add(1)（甚至同一worker多次Add），导致WaitGroup计数远大于实际启动的goroutine数量。
• 启动的goroutine与Done()调用不匹配，部分goroutine未正确调用WaitGroup.Done()，主goroutine永远无法完成等待。

2. 未消费通道导致阻塞 #

• outputChan和errorChan仅被worker发送数据，但无任何goroutine接收，通道填满后worker会因发送操作阻塞，无法退出。
• 第一个worker的goroutine依赖done通道退出，但done通道从未被发送信号或关闭，导致该goroutine一直阻塞在select逻辑中。

3. 重复启动Worker逻辑混乱 #

• 同一个worker被启动两个goroutine：一个执行filterRows并直接写输出，另一个执行applyFilter并发送到outputChan，两个goroutine同时读取inputChan，逻辑重叠冲突。

4. CSV分块与文件读取逻辑错误 #

• 先创建csv.Reader但后续直接调用inputFile.Read()读取字节，导致文件指针位置混乱，无法正确读取数据。
• 按固定字节大小分块会截断CSV行，导致解析失败；且生成的chunk未实际发送到inputChan，worker无数据可处理。

解决方案 #

1. 修复WaitGroup使用 #

• 仅为每个实际启动的goroutine调用WaitGroup.Add(1)，在goroutine内部用defer wg.Done()确保计数正确递减。
• 删除不必要的WaitGroup初始化和重复Add操作。

2. 处理未消费通道 #

• 移除冗余的outputChan，直接在worker中完成过滤并写入输出（需加锁保证writer线程安全）。
• 启动专门的错误处理goroutine，消费errorChan中的错误，避免worker因发送错误阻塞。

3. 清理重复Worker逻辑 #

• 合并worker的处理逻辑，每个worker仅启动一个goroutine，负责接收chunk、执行过滤、处理输出。
• 移除无效的done通道，依赖关闭inputChan让worker的for range循环自动退出。

4. 修复CSV分块与读取 #

• 按行读取CSV文件，积累到指定行数后组成chunk，避免截断行。
• 将生成的chunk实际发送到inputChan，再关闭通道。

修正后的关键代码片段 #

简化核心结构与Worker逻辑 #

type Splitter struct {
    workers        int
    inputFilePath  string
    outputFilePath string
    chunkSize      int64 // 按行数量分块，而非字节
    mu             sync.Mutex
}

type worker struct {
    id         int
    inputChan  <-chan [][]string
    writer     *csv.Writer
    errorChan  chan<- error
    wg         *sync.WaitGroup
}

func (w *worker) run() {
    defer w.wg.Done()
    for rows := range w.inputChan {
        filteredRows := make([][]string, 0)
        for _, row := range rows {
            // 执行过滤逻辑：保留第二列数值大于10的行
            if len(row) >= 2 {
                value, err := strconv.Atoi(row[1])
                if err != nil {
                    w.errorChan <- fmt.Errorf("worker %d: parse error: %v", w.id, err)
                    continue
                }
                if value > 10 {
                    filteredRows = append(filteredRows, row)
                }
            }
        }
        // 写输出时加锁保证线程安全
        w.writer.Lock()
        w.writer.WriteAll(filteredRows)
        w.writer.Unlock()
    }
}

func (s *Splitter) Split() error {
    inputFile, err := os.Open(s.inputFilePath)
    if err != nil {
        return fmt.Errorf("open input file: %v", err)
    }
    defer inputFile.Close()

    reader := csv.NewReader(inputFile)
    reader.FieldsPerRecord = -1
    reader.TrimLeadingSpace = true

    outputFile, err := os.Create(s.outputFilePath)
    if err != nil {
        return fmt.Errorf("create output file: %v", err)
    }
    defer outputFile.Close()

    // 给CSV writer添加锁，保证多worker写入安全
    type lockedWriter struct {
        *csv.Writer
        sync.Mutex
    }
    writer := &lockedWriter{csv.NewWriter(outputFile), sync.Mutex{}}

    inputChan := make(chan [][]string, s.workers)
    errorChan := make(chan error, s.workers)
    var wg sync.WaitGroup

    // 启动workers
    for i := 0; i < s.workers; i++ {
        wg.Add(1)
        w := worker{
            id:         i,
            inputChan:  inputChan,
            writer:     writer.Writer,
            errorChan:  errorChan,
            wg:         &wg,
        }
        go w.run()
    }

    // 后台处理worker错误
    go func() {
        for err := range errorChan {
            log.Printf("worker error: %v", err)
        }
    }()

    // 按行读取并分块发送给workers
    var currentChunk [][]string
    for {
        row, err := reader.Read()
        if err == io.EOF {
            break
        }
        if err != nil {
            return fmt.Errorf("read csv row: %v", err)
        }
        currentChunk = append(currentChunk, row)
        if int64(len(currentChunk)) >= s.chunkSize {
            inputChan <- currentChunk
            currentChunk = make([][]string, 0)
        }
    }
    // 发送最后一块数据
    if len(currentChunk) > 0 {
        inputChan <- currentChunk
    }

    close(inputChan)
    wg.Wait()
    close(errorChan)

    // 刷新writer缓冲区
    writer.Lock()
    writer.Flush()
    writer.Unlock()

    return writer.Error()
}

主函数调用 #

func main() {
    s := &Splitter{
        workers:        4,
        inputFilePath:  "input.csv",
        outputFilePath: "output.csv",
        chunkSize:      100, // 每块处理100行
    }
    if err := s.Split(); err != nil {
        log.Fatal(err)
    }
}

内容的提问来源于stack exchange，提问作者algo tourist