Go语言中并行计算独立词汇数量的策略与实现

package main

import (
    "bufio"
    "fmt"
    "io"
    "os"
    "regexp"
    "strings"
    "sync"
)

// wordRegexp 用于匹配字母和数字组成的词汇
var wordRegexp = regexp.MustCompile(`[a-zA-Z0-9]+`)

// splitter Goroutine：从 reader 读取文本，按行发送到 inputChan
func splitter(reader io.Reader, inputChan chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(inputChan) // 确保在 splitter 完成后关闭 inputChan

    scanner := bufio.NewScanner(reader)
    for scanner.Scan() {
        inputChan <- scanner.Text() // 将每一行作为文本块发送
    }
    if err := scanner.Err(); err != nil {
        fmt.Fprintf(os.Stderr, "Error reading input: %v\n", err)
    }
}

// worker Goroutine：从 inputChan 接收文本块，处理并统计局部独立词汇，发送到 outputChan
func worker(inputChan <-chan string, outputChan chan<- map[string]struct{}, wg *sync.WaitGroup) {
    defer wg.Done()

    localDistinctWords := make(map[string]struct{})
    for line := range inputChan { // inputChan 关闭时，此循环会自动结束
        // 提取词汇并规范化：转小写，去除标点
        words := wordRegexp.FindAllString(strings.ToLower(line), -1)
        for _, word := range words {
            localDistinctWords[word] = struct{}{} // 存入局部集合
        }
    }
    // 将局部结果发送给聚合器
    outputChan <- localDistinctWords
}

// aggregator Goroutine：从 outputChan 接收局部词汇集，合并到全局词汇集
func aggregator(outputChan <-chan map[string]struct{}, globalDistinctWords *sync.Map, wg *sync.WaitGroup) {
    defer wg.Done()

    for localWords := range outputChan { // outputChan 关闭时，此循环会自动结束
        for word := range localWords {
            globalDistinctWords.Store(word, struct{}{}) // sync.Map 是并发安全的
        }
    }
}

func main() {
    numWorkers := 4 // 工作者数量，可根据CPU核心数或实际负载调整
    var splitterWg sync.WaitGroup
    var workerWg sync.WaitGroup
    var aggregatorWg sync.WaitGroup

    // 定义通道：
    // inputChan 用于 splitter 到 workers 传递文本行
    inputChan := make(chan string, 100)
    // outputChan 用于 workers 到 aggregator 传递局部独立词汇集合
    outputChan := make(chan map[string]struct{}, numWorkers) // 缓冲区大小至少为 numWorkers，以避免阻塞

    // globalDistinctWords 使用 sync.Map 保证并发安全地存储全局独立词汇
    globalDistinctWords := &sync.Map{}

    // 1. 启动 splitter Goroutine
    splitterWg.Add(1)
    go splitter(os.Stdin, inputChan, &splitterWg)

    // 2. 启动多个 worker Goroutine
    workerWg.Add(numWorkers)
    for i := 0; i < numWorkers; i++ {
        go worker(inputChan, outputChan, &workerWg)
    }

    // 3. 启动 aggregator Goroutine
    aggregatorWg.Add(1)
    go aggregator(outputChan, globalDistinctWords, &aggregatorWg)

    // 协调 Goroutine 的生命周期：

    // 等待 splitter 完成其工作。当 splitter 完成后，inputChan 会被关闭，通知所有 worker 停止接收。
    splitterWg.Wait()

    // 等待所有 worker 完成其工作