挑战与并行化需求
统计文本中独立词汇的数量是一个常见的编程挑战。当面对大量文本数据时，单线程处理效率低下，因此引入并行编程成为必然选择。go语言以其轻量级并发原语——goroutine和channel——为实现高效并行解决方案提供了天然优势。本教程将详细阐述如何利用go语言构建一个并行化的独立词汇计数程序。
核心概念：Map/Reduce范式
独立词汇计数问题天然契合Map/Reduce范式。该范式将复杂任务分解为两个主要阶段：
- Map（映射）阶段： 将输入数据分割成小块，每个工作单元独立处理一个数据块，生成一组中间结果。在本例中，每个工作者（Worker）负责处理一个文本片段，提取其中所有的独立词汇，并生成该片段的局部独立词汇集合。
- Reduce（归约）阶段： 收集所有Map阶段生成的中间结果，并对其进行聚合和合并，最终得出全局的最终结果。在本例中，一个聚合器（Aggregator）收集所有工作者生成的局部独立词汇集合，并将其合并为一个全局的独立词汇集合，最终计算出总数。
并行架构设计
基于Map/Reduce范式，我们可以设计一个三层架构的并行词汇计数系统：
-
数据切分器 (Splitter)：
- 职责： 负责从标准输入（或其他数据源）读取原始文本数据。
- 功能： 将连续的文本流切分为更小的、可独立处理的文本块（例如，按行或按固定字节数）。
- 输出： 将这些文本块通过Go Channel发送给工作者。
-
工作者 (Workers)：
立即学习“go语言免费学习笔记（深入）”；
- 职责： 从切分器接收文本块，并进行并行处理。
- 功能： 对每个接收到的文本块执行词汇提取、规范化（如转换为小写、去除标点符号）和局部独立词汇统计。
- 输出： 将每个工作者统计出的局部独立词汇集合（通常是一个 map[string]struct{}）通过另一个Go Channel发送给聚合器。
- 特点： 多个工作者并发运行，共享同一个输入通道，从而实现负载均衡。
-
结果聚合器 (Aggregator)：
- 职责： 从所有工作者接收局部独立词汇集合，并进行最终的合并。
- 功能： 维护一个全局的独立词汇集合，将从工作者接收到的局部集合中的词汇逐一添加到全局集合中。由于聚合器会接收来自多个工作者的并发写入，因此需要确保其内部数据结构是并发安全的。
- 输出： 最终的全局独立词汇总数。
通信与协调机制：
- Go Channels： 作为数据流动的管道，连接Splitter、Workers和Aggregator。inputChan 用于Splitter向Workers发送文本块，outputChan 用于Workers向Aggregator发送局部结果。
- sync.WaitGroup： 用于协调Goroutine的生命周期。Splitter、Workers和Aggregator在完成各自任务后，会通知 WaitGroup，主Goroutine通过等待 WaitGroup 来确保所有任务都已完成。
Go语言实现细节
以下是一个基于上述架构的Go语言示例代码结构，展示了如何使用Goroutine和Channel实现并行词汇计数。
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 完成其工作 
