Go语言固定100个goroutine并行MD5哈希手机号问题求助

嘿，我来帮你搞定这个问题！你之前尝试用Pipeline和有界并行没成功，大概率是在goroutine数量控制、通道关闭时机或者任务/结果的同步上踩了坑。下面我给你一个清晰可行的实现方案，一步步拆解给你看。

核心思路 #

我们要实现的是固定数量的Worker池：

1. 启动100个Worker Goroutine，它们会一直等待任务；
2. 从文件逐行读取手机号，作为任务发送到任务通道；
3. Worker从任务通道取手机号，计算MD5后把结果发送到结果通道；
4. 主Goroutine收集所有结果，直到所有Worker完成任务。

完整实现代码 #

package main

import (
	"bufio"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"os"
	"sync"
)

// Task 定义单个处理任务：待哈希的手机号
type Task struct {
	Phone string
}

// Result 定义处理结果：手机号+对应的MD5哈希值
type Result struct {
	Phone string
	MD5   string
}

// worker 每个Worker的逻辑：从任务通道取数据，计算MD5，发送结果到结果通道
func worker(id int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done() // Worker完成后通知WaitGroup

	for task := range tasks {
		// 计算MD5哈希
		hashBytes := md5.Sum([]byte(task.Phone))
		md5Str := hex.EncodeToString(hashBytes[:])
		// 发送结果到通道
		results <- Result{Phone: task.Phone, MD5: md5Str}
		fmt.Printf("Worker %d 处理完成：%s\n", id, task.Phone)
	}
}

func main() {
	const workerCount = 100 // 固定100个Worker
	tasks := make(chan Task, 100)  // 带缓冲的任务通道，避免文件读取阻塞
	results := make(chan Result, 100) // 带缓冲的结果通道
	var wg sync.WaitGroup

	// 启动所有Worker
	for i := 1; i <= workerCount; i++ {
		wg.Add(1)
		go worker(i, tasks, results, &wg)
	}

	// 单独开一个Goroutine读取文件并发送任务
	go func() {
		file, err := os.Open("phones.txt") // 替换成你的手机号文件路径
		if err != nil {
			fmt.Printf("打开文件失败：%v\n", err)
			return
		}
		defer file.Close()
		defer close(tasks) // 所有任务发送完后关闭任务通道，Worker会自动退出循环

		scanner := bufio.NewScanner(file)
		for scanner.Scan() {
			phone := scanner.Text()
			if phone != "" { // 跳过空行
				tasks <- Task{Phone: phone}
			}
		}

		if err := scanner.Err(); err != nil {
			fmt.Printf("读取文件出错：%v\n", err)
		}
	}()

	// 单独开一个Goroutine等待所有Worker完成，然后关闭结果通道
	go func() {
		wg.Wait()
		close(results)
	}()

	// 主Goroutine收集并处理结果（这里可以改成写入文件等逻辑）
	for res := range results {
		fmt.Printf("手机号：%s，MD5值：%s\n", res.Phone, res.MD5)
	}

	fmt.Println("所有任务处理完成！")
}

关键细节解释 #

1. 固定Worker数量：通过workerCount常量严格控制启动100个Goroutine，不会出现无限制创建的情况；
2. 带缓冲通道：任务和结果通道设置缓冲（这里设为100），可以避免文件读取或结果处理时的阻塞，提升并发效率；
3. WaitGroup的作用：用来追踪所有Worker的状态，确保所有任务都处理完成后再关闭结果通道；
4. 通道关闭时机：
   • 文件读取完成后关闭任务通道，Worker会在通道为空且关闭时自动退出循环；
   • 所有Worker完成后关闭结果通道，主Goroutine就能知道所有结果都已处理完毕，不会一直阻塞。

你之前可能踩的坑 #

• 没有正确关闭通道，导致Goroutine泄漏；
• 没有用WaitGroup等待所有Worker完成，结果通道提前关闭导致panic；
• Worker数量没有固定，比如循环读取每行都创建一个Goroutine，导致数量失控。

内容的提问来源于stack exchange，提问作者Jau L