Go并发Grep实现报错：循环启动Goroutine时发送至已关闭通道

问题根源 #

1. 通道过早关闭：searchPaths函数中，启动所有goroutine后立即关闭out通道，此时goroutine还未开始向通道发送数据，后续goroutine执行发送操作时会触发Send on closed channel panic。
2. 循环变量共享问题：goroutine直接引用循环变量path，Go的循环变量在迭代中是复用的，导致多个goroutine可能使用同一个path值，搜索错误的文件。
3. WaitGroup与通道遍历顺序错误：main函数中先调用wg.Wait()再遍历通道，若通道未关闭会引发死锁；而当前场景下通道被过早关闭，又导致发送panic。

修复方案 #

1. 延迟关闭通道：新增一个goroutine，等待所有搜索goroutine执行完毕后再关闭输出通道，保证所有发送操作完成后才关闭通道。
2. 传递循环变量副本：启动goroutine时将当前path作为参数传入，让每个goroutine持有独立的路径副本。
3. 调整main流程：直接遍历输出通道，通道关闭后遍历会自动终止，无需提前等待WaitGroup。

修改后的完整代码 #

package main

import (
	"fmt"
	"io/fs"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
)

type SearchResult struct {
	line       string
	lineNumber int
}

type Display struct {
	filePath string
	SearchResult
}

var wg sync.WaitGroup

func (d Display) PrettyPrint() {
	fmt.Printf("Line Number: %v\nFilePath: %v\nLine: %v\n\n", d.lineNumber, d.filePath, d.line)
}

func searchLine(pattern string, line string, lineNumber int) (SearchResult, bool) {
	if strings.Contains(line, pattern) {
		return SearchResult{lineNumber: lineNumber + 1, line: line}, true
	}
	return SearchResult{}, false
}

func splitIntoLines(file string) []string {
	lines := strings.Split(file, "\n")
	return lines
}

func fileFromPath(path string) string {
	fileContent, err := ioutil.ReadFile(path)

	if err != nil {
		log.Fatal(err)
	}

	return string(fileContent)
}

func getRecursiveFilePaths(inputDir string) []string {
	var paths []string
	err := filepath.Walk(inputDir, func(path string, info fs.FileInfo, err error) error {
		if err != nil {
			fmt.Printf("prevent panic by handling failure accessing a path %q: %v\n", path, err)
			return err
		}
		if !info.IsDir() {
			paths = append(paths, path)
		}
		return nil
	})
	if err != nil {
		fmt.Printf("Error walking the path %q: %v\n", inputDir, err)
	}
	return paths
}

func searchPaths(paths []string, pattern string) <-chan Display {
	out := make(chan Display)

	// 单独启动goroutine等待所有搜索任务完成后关闭通道
	go func() {
		wg.Wait()
		close(out)
	}()

	for _, path := range paths {
		wg.Add(1)
		// 将当前path作为参数传入goroutine，避免共享循环变量
		go func(p string) {
			defer wg.Done()
			for _, display := range searchFile(p, pattern) {
				out <- display
			}
		}(path)
	}

	return out
}

func searchFile(path string, pattern string) []Display {
	var out []Display
	input := fileFromPath(path)
	lines := splitIntoLines(input)
	for index, line := range lines {
		if searchResult, ok := searchLine(pattern, line, index); ok {
			out = append(out, Display{path, searchResult})
		}
	}
	return out
}

func main() {
	if len(os.Args) < 3 {
		log.Fatal("Usage: ./concurrent-grep <pattern> <directory>")
	}
	pattern := os.Args[1]
	dirPath := os.Args[2]

	paths := getRecursiveFilePaths(dirPath)

	out := searchPaths(paths, pattern)
	// 直接遍历通道，通道关闭后循环自动结束
	for d := range out {
		d.PrettyPrint()
	}
}

关键修改说明 #

• 在searchPaths中新增了一个负责等待WaitGroup并关闭通道的goroutine，确保所有发送操作完成后才关闭通道。
• 启动搜索goroutine时，将循环变量path作为参数传递给匿名函数，避免多个goroutine共享同一个变量实例。
• main函数中移除了wg.Wait()，直接遍历输出通道，利用通道关闭后遍历自动终止的特性完成结果接收。

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