Go并发编程中的常见陷阱与性能优化

package main

import (
    "bufio"
    "crypto/sha1"
    "fmt"
    "io"
    "log"
    "os"
    "strings"
    "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 // 添加互斥锁保护recordStrings
}

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)
    for scanner.Scan() { //Scan the file
        retText := scanner.Text()
        if strings.HasPrefix(retText, "//") {
            wg.Add(1)
            // 传递record的副本，避免在goroutine内部修改外部的record slice
            // 或者在handleRecord内部处理record slice的拷贝
            go u.handleRecord(append([]string{}, record...), wg)
            record = []string{}
        } else {
            record = append(record, retText)
        }
    }
    // 处理文件末尾可能未处理的最后一个记录
    if len(record) > 0 {
        wg.Add(1)
        go u.handleRecord(record, wg)
    }
    file.Close()
    wg.Wait()
    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)

    // 保护对recordStrings的写入
    u.mu.Lock()
    u.recordStrings[t] = recString
    u.mu.Unlock()

    // 通道操作本身是并发安全的，不需要额外锁
    u.recordUnits <- unit{tag: t}
}

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)
    }
}

// ... (之前的代码保持不变，包括添加的sync.Mutex)

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通道
    // 确保在所有生产者都完成后关闭通道，或者使用context取消
    go u.consumeRecords() // 启动消费者

    for scanner.Scan() { //Scan the file
        retText := scanner.Text()
        if strings.HasPrefix(retText, "//") {
            wg.Add(1)
            go u.handleRecord(append([]string{}, record...), wg)
            record = []string{}
        } else {
            record = append(record, retText)
        }
    }
    // 处理文件末尾可能未处理的最后一个记录
    if len(record) > 0 {
        wg.Add(1)
        go u.handleRecord(record, wg)
    }
    file.Close()
    wg.Wait() // 等待所有handleRecord goroutine完成

    close(u.recordUnits) // 所有生产者完成后关闭通道，通知消费者
    // 消费者goroutine也需要等待，但这里不直接等待，而是通过range循环自动退出
    // 如果需要明确等待消费者完成，需要为消费者添加另一个WaitGroup
    t1 := time.Now()
    fmt.Println(t1.Sub(t0))
}

// 新增的消费者goroutine函数
func (u *uniprot) consumeRecords() {
    for r := range u.recordUnits {
        // 在这里处理从通道中取出的unit数据
        // 例如：存储到数据库、进行进一步分析等
        _ = r // 避免未使用变量的警告，实际应用中会处理r
        // fmt.Printf("Consumed record tag: %s\n", r.tag)
    }
    // 通道关闭后，range循环会自动退出
    // fmt.Println("Record consumer finished.")
}

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

// ... (uniprot结构体可以考虑将recordStrings的值类型改为[]byte)
// type uniprot struct {
//     // ...
//     recordStrings map[string][]byte
// }

func (u *uniprot) handleRecord(record []string, wg *sync.WaitGroup) {
    defer wg.Done()

    // 使用bytes.Buffer高效拼接，避免多次字符串拷贝
    var buf strings.Builder
    for i, line := range record {
        buf.WriteString(line)
        if i < len(record)-1 {
            buf.WriteString("\n")
        }
    }
    recString := buf.String() // 最终转换为string

    // 如果hashfunc可以接受[]byte，则直接传入[]byte
    // recBytes := []byte(recString) // 或直接从buf.Bytes()获取
    // t := hashfuncBytes(recBytes) // 假设存在一个处理[]byte的hashfunc

    t := hashfunc(recString)

    u.mu.Lock()
    u.recordStrings[t] = recString // 如果recordStrings改为map[string][]byte，这里也需要相应修改
    u.mu.Unlock()

    u.recordUnits <- unit{tag: t}
}

// 如果hashfunc可以处理[]byte，可以这样修改
// func hashfuncBytes(record []byte) (hashtag string) {
//     hash := sha1.New()
//     hash.Write(record) // 直接写入[]byte
//     hashtag = string(hash.Sum(nil))
//     return
// }