Go并发编程：使用sync.WaitGroup安全管理通道与解决数据不一致问题

package main

import (
    "encoding/csv"
    "fmt"
    "io"
    "log"
    "os"
    "regexp"
    "sync" // 引入sync包
)

var (
    cleanRe         *regexp.Regexp = regexp.MustCompile("[^0-9]+")
    comma           rune           = '\t'
    fieldsPerRecord                = -1
)

// clean 函数用于清洗字符串，移除所有非数字字符，并检查长度。
func clean(s string) string {
    clean := cleanRe.ReplaceAllLiteralString(s, "")
    if len(clean) < 6 {
        return ""
    }
    return clean
}

// uniqueChannel 是消费者协程，从inputChan接收数据并进行去重处理后打印。
func uniqueChannel(inputChan chan []string) {
    uniq := make(map[string]map[string]bool)
    i := 0
    // 遍历inputChan直到其关闭且所有数据被取出。
    for record := range inputChan {
        i++
        id, v := record[0], record[1]
        if uniq[id] == nil {
            uniq[id] = make(map[string]bool)
        }
        // 只有当id-v组合首次出现时才记录并打印。
        if !uniq[id][v] {
            uniq[id][v] = true
            fmt.Println(id, string(comma), v)
        }
    }
    log.Println("digest ", i)
}

// processFile 是生产者协程，负责处理单个文件并将清洗后的记录发送到outputChan。
func processFile(fileName string, outputChan chan []string) {
    f, err := os.Open(fileName)
    if err != nil {
        log.Fatal(err)
    }
    defer f.Close() // 确保文件句柄在函数返回前关闭。

    r := csv.NewReader(f)
    r.FieldsPerRecord = fieldsPerRecord
    r.Comma = comma

    // 循环读取文件中的记录。
    for record, err := r.Read(); err != io.EOF; record, err = r.Read() {
        if err != nil {
            // 忽略读取错误，继续处理下一个记录。
            continue
        }
        id := record[0]
        // 处理记录中的每个值。
        for _, v := range record[1:] {
            if cleanV := clean(v); cleanV != "" {
                outputChan <- []string{id, cleanV} // 将清洗后的值发送到通道。
            }
        }
    }
}

func main() {
    // 示例输入文件列表。请确保ex.tsv文件存在或替换为实际文件。
    inputs := []string{"ex.tsv"}
    recordChan := make(chan []string) // 创建一个无缓冲通道用于传递数据。

    var wg sync.WaitGroup // 声明一个WaitGroup用于同步协程。

    // 启动生产者协程，处理每个输入文件。
    for _, fName := range inputs {
        wg.Add(1) // 每启动一个文件处理协程，WaitGroup计数器加1。
        go func(fname string) { // 使用闭包捕获fname，避免变量在循环中被覆盖。
            defer wg.Done() // 协程结束时（无论正常退出或panic），调用Done()使计数器减1。
            processFile(fname, recordChan)
        }(fName)
    }

    // 启动一个独立的协程，等待所有生产者协程完成，然后关闭recordChan。
    go func() {
        wg.Wait()          // 阻塞直到所有通过Add()注册的协程都调用了Done()。
        close(recordChan)  // 所有生产者都完成后，安全关闭recordChan。
    }()

    // 启动消费者协程，处理recordChan中的数据。
    uniqueChannel(recordChan) // uniqueChannel会一直运行直到recordChan被关闭且所有数据被取出。

    log.Println("所有任务完成。")
}