Go语言中Scanner.Bytes()传入通道的多Worker数据处理问题

最近我折腾了这么个需求：生成格式为[随机字符串]|[随机字符串]的内容（两边的随机串完全一致），然后用16个Worker线程来处理这些字符串，最终验证两侧内容没有因为多线程传输而损坏。下面把实现思路和关键代码分享给大家：

整体流程 #

整个流程分为三步，清晰明了：

1. 读取输入到通道：用Scanner.Bytes()把生成的字符串读取到输入通道，这里要注意字节数组的复制，避免Scanner复用内存导致数据覆盖
2. Worker线程处理：启动16个Worker，从输入通道拿到字节数组转成字符串，再传到输出通道
3. 验证一致性：从输出通道取出字符串，按|分割后对比两侧内容，确保数据没损坏

关键代码片段 #

核心对称字符串生成逻辑 #

这部分负责生成foo|foo格式的字符串，两边是相同的随机小写字母串：

package main

import (
	"bufio"
	"log"
	"math/rand"
	"strings"
	"time"
)

func generatePair() string {
	rand.Seed(time.Now().UnixNano())
	// 生成3位随机小写字母
	b := make([]byte, 3)
	for i := range b {
		b[i] = 'a' + byte(rand.Intn(26))
	}
	s := string(b)
	return s + "|" + s
}

扫描输入并发送到通道 #

这部分是把生成的内容读取到输入通道的核心代码，注意必须复制字节数组，因为Scanner的Bytes()返回的是内部缓冲区的引用，后续扫描会覆盖：

// Send data to input channel using Scanner.Bytes()
func sendInput(scanner *bufio.Scanner, inputChan chan<- []byte) {
	defer close(inputChan)
	for scanner.Scan() {
		// 复制字节数组，避免Scanner后续操作覆盖数据
		inputBytes := make([]byte, len(scanner.Bytes()))
		copy(inputBytes, scanner.Bytes())
		inputChan <- inputBytes
	}
	// 处理扫描过程中的错误
	if err := scanner.Err(); err != nil {
		log.Fatalf("Failed to scan input: %v", err)
	}
}

Worker线程实现 #

每个Worker的逻辑很简单，就是把字节数组转成字符串，然后传到输出通道：

func worker(inputChan <-chan []byte, outputChan chan<- string) {
	for data := range inputChan {
		outputChan <- string(data)
	}
}

最终验证逻辑 #

从输出通道取出字符串，分割后对比两侧内容，检查是否损坏：

func validate(outputChan <-chan string) {
	for str := range outputChan {
		parts := strings.Split(str, "|")
		if len(parts) != 2 || parts[0] != parts[1] {
			log.Printf("Data corruption detected: %s", str)
		} else {
			log.Printf("Valid pair: %s", str)
		}
	}
}

主函数入口（完整调用示例） #

把各个部分串起来的主函数：

func main() {
	inputChan := make(chan []byte, 100)
	outputChan := make(chan string, 100)

	// 启动16个Worker线程
	for i := 0; i < 16; i++ {
		go worker(inputChan, outputChan)
	}

	// 启动验证goroutine
	go func() {
		validate(outputChan)
	}()

	// 生成测试数据并发送到输入通道
	// 实际场景中可以替换为bufio.NewScanner(os.Stdin)读取外部输入
	for i := 0; i < 1000; i++ {
		pair := generatePair()
		inputChan <- []byte(pair)
	}
	close(inputChan)

	// 等待所有处理完成
	time.Sleep(1 * time.Second)
}

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