Go并发编程中的常见陷阱与高效实践

package main

import (
    "bufio"
    "crypto/sha1"
    "fmt"
    "io"
    "log"
    "os"
    "strings"
    "sync" // 引入 sync 包
    "time"
)

type producer struct {
    parser uniprot
}

type unit struct {
    tag string
}

type uniprot struct {
    filenames     []string
    recordUnits   chan unit
    recordStrings map[string]string
    mu            sync.Mutex // 添加互斥锁
}

func main() {
    p := producer{parser: uniprot{}}
    p.parser.recordUnits = make(chan unit, 1000000)
    p.parser.recordStrings = make(map[string]string)
    p.parser.collectRecords(os.Args[1])
}

func (u *uniprot) collectRecords(name string) {
    fmt.Println("file to open ", name)
    t0 := time.Now()
    wg := new(sync.WaitGroup)
    record := []string{}
    file, err := os.Open(name)
    errorCheck(err)
    scanner := bufio.NewScanner(file)

    // 启动一个goroutine来消费recordUnits通道中的数据，防止死锁
    // 这是一个简化的消费示例，实际应用中可能需要更复杂的处理逻辑
    go func() {
        for range u.recordUnits {
            // 这里可以对取出的unit进行处理，例如存储到数据库或进一步分析
            // 如果仅仅是为了防止死锁，空循环也可以
        }
    }()

    for scanner.Scan() { //Scan the file
        retText := scanner.Text()
        if strings.HasPrefix(retText, "//") {
            wg.Add(1)
            // 传递 record 的副本，避免在 goroutine 内部修改原始 slice
            recordCopy := make([]string, len(record))
            copy(recordCopy, record)
            go u.handleRecord(recordCopy, wg)
            record = []string{} // 重置 record
        } else {
            record = append(record, retText)
        }
    }
    // 处理文件末尾可能剩余的 record
    if len(record) > 0 {
        wg.Add(1)
        recordCopy := make([]string, len(record))
        copy(recordCopy, record)
        go u.handleRecord(recordCopy, wg)
    }

    file.Close()
    wg.Wait()
    close(u.recordUnits) // 在所有生产者完成后关闭通道
    t1 := time.Now()
    fmt.Println(t1.Sub(t0))
}

func (u *uniprot) handleRecord(record []string, wg *sync.WaitGroup) {
    defer wg.Done()
    recString := strings.Join(record, "\n")
    t := hashfunc(recString)

    u.recordUnits <- unit{tag: t} // 写入通道，此处仍需消费者

    // 使用互斥锁保护对 recordStrings 的写入
    u.mu.Lock()
    u.recordStrings[t] = recString
    u.mu.Unlock()
}

func hashfunc(record string) (hashtag string) {
    hash := sha1.New()
    io.WriteString(hash, record)
    hashtag = string(hash.Sum(nil))
    return
}

func errorCheck(err error) {
    if err != nil {
        log.Fatal(err)
    }
}

// ... (之前的代码，包括 uniprot 结构体和 main 函数的修改) ...

func (u *uniprot) collectRecords(name string) {
    fmt.Println("file to open ", name)
    t0 := time.Now()
    wg := new(sync.WaitGroup)
    record := []string{}
    file, err := os.Open(name)
    errorCheck(err)
    scanner := bufio.NewScanner(file)

    // 启动一个独立的 goroutine 来消费 recordUnits 通道中的数据
    // 这是一个简化的消费者，实际应用中会处理这些数据
    go func() {
        for rUnit := range u.recordUnits {
            // 这里可以对 rUnit 进行进一步处理，例如写入数据库、进行分析等
            _ = rUnit // 避免 unused variable 警告
        }
        // 当通道关闭且所有数据被取出后，此goroutine会退出
    }()

    for scanner.Scan() { //Scan the file
        retText := scanner.Text()
        if strings.HasPrefix(retText, "//") {
            wg.Add(1)
            recordCopy := make([]string, len(record))
            copy(recordCopy, record)
            go u.handleRecord(recordCopy, wg)
            record = []string{}
        } else {
            record = append(record, retText)
        }
    }
    if len(record) > 0 { // 处理文件末尾可能剩余的 record
        wg.Add(1)
        recordCopy := make([]string, len(record))
        copy(recordCopy, record)
        go u.handleRecord(recordCopy, wg)
    }

    file.Close()
    wg.Wait()          // 等待所有 handleRecord goroutine 完成
    close(u.recordUnits) // 所有生产者完成后，关闭通道，通知消费者可以退出了
    t1 := time.Now()
    fmt.Println(t1.Sub(t0))
}

// ... (handleRecord 和 hashfunc, errorCheck 函数保持不变) ...

func hashfunc(record string) (hashtag string) {
    hash := sha1.New()
    // 直接使用 []byte 写入，避免 string 到 []byte 的隐式转换和复制
    hash.Write([]byte(record)) // 这里的 []byte(record) 仍会创建副本，但如果 record 本身就是 []byte 则可避免
    hashtag = string(hash.Sum(nil))
    return
}