Go并发编程：避免Goroutine开销陷阱与常见并发问题解析

u.recordStrings[t] = recString

import "sync"

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

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

    // 在写入map之前加锁
    u.mu.Lock()
    u.recordStrings[t] = recString
    u.mu.Unlock() // 写入完成后解锁

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

p.parser.recordUnits = make(chan unit, 1000000)

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作为通道消费者
    go func() {
        for range u.recordUnits {
            // 在实际应用中，这里会处理从通道中接收到的unit数据
            // 例如：存储到数据库、进行后续分析等
            // 在此示例中，我们仅为了避免死锁而简单地消费掉数据
        }
    }()

    for scanner.Scan() { // 扫描文件
        retText := scanner.Text()
        if strings.HasPrefix(retText, "//") {
            wg.Add(1)
            go u.handleRecord(record, wg)
            record = []string{}
        } else {
            record = append(record, retText)
        }
    }
    file.Close()

    wg.Wait() // 等待所有handleRecord Goroutine完成
    close(u.recordUnits) // 所有发送者完成后，关闭通道
    // 关闭通道后，上面启动的消费者Goroutine的 for range 循环将自动结束并退出。

    t1 := time.Now()
    fmt.Println(t1.Sub(t0))
}

// 示例：将record改为[][]byte
func (u *uniprot) handleRecord(record [][]byte, wg *sync.WaitGroup) {
    defer wg.Done()
    // 假设hashfunc也接受[]byte
    // recBytes := bytes.Join(record, []byte("\n")) // 需要导入bytes包
    // t := hashfuncBytes(recBytes) // 假设有hashfuncBytes处理[]byte

    // ... 其他逻辑
}

package main

import (
    "bufio"
    "crypto/sha1"
    "fmt"
    "io"
    "log"
    "os"
    "runtime" // 导入runtime包
    "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 // 添加互斥锁保护map
}

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU()) // 设置GOMAXPROCS为CPU核心数

    p := producer{parser: uniprot{}}
    // 缓冲区大小根据实际情况调整，过大会占用大量内存，过小会频繁阻塞
    p.parser.recordUnits = make(chan unit, 100000) // 适当减小缓冲区大小
    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{} // 仍然使用[]string，但注意其性能影响
    file, err := os.Open(name)
    errorCheck(err)
    scanner := bufio.NewScanner(file)

    // 启动通道消费者Goroutine
    // 消费者Goroutine也应加入WaitGroup，以确保其在程序结束前完成
    wg.Add(1) // 为消费者Goroutine增加计数
    go func() {
        defer wg.Done() // 消费者完成时调用Done
        for range u.recordUnits {
            // 在这里处理从通道中接收到的unit数据
            //