在Go语言的实际开发中，处理体积较大的文本文件时，单线程逐行读取的方式往往会消耗较多时间，利用并发特性可以显著提升读取效率，但并发场景下容易出现数据竞争、资源争抢等问题，需要合理的设计保障读取过程的安全。
并发读取的核心思路
并发读取文本文件的核心逻辑是将文件拆分为多个独立的读取单元，每个单元由单独的goroutine处理，最后将各单元的结果汇总。常见的拆分方式有两种，一种是按文件字节偏移量拆分，另一种是按行数拆分，前者更适合大文件场景，后者实现逻辑更简单。
按字节偏移量拆分的实现
首先需要获取文件的总大小，然后根据并发数计算每个goroutine需要读取的字节区间，每个goroutine从指定的偏移量开始读取对应长度的内容，最后处理可能出现的不完整行问题。
package main
import (
"bufio"
"fmt"
"io"
"os"
"sync"
)
// 读取文件的单个区间
func readSection(file *os.File, offset int64, length int64, resultChan chan<- string, wg *sync.WaitGroup) {
defer wg.Done()
// 移动文件指针到指定偏移量
_, err := file.Seek(offset, io.SeekStart)
if err != nil {
fmt.Printf("移动文件指针失败: %vn", err)
return
}
reader := bufio.NewReader(file)
// 读取指定长度的内容
buffer := make([]byte, length)
n, err := reader.Read(buffer)
if err != nil && err != io.EOF {
fmt.Printf("读取内容失败: %vn", err)
return
}
// 处理可能的不完整行，避免拆分到行中间
content := string(buffer[:n])
// 如果当前区间不是文件末尾，且最后一个字符不是换行符，读取到下一个换行符为止
if offset+length < getFileSize(file) {
remaining, _ := reader.ReadString('n')
content += remaining
}
resultChan <- content
}
// 获取文件大小
func getFileSize(file *os.File) int64 {
stat, err := file.Stat()
if err != nil {
return 0
}
return stat.Size()
}
func main() {
// 打开目标文件
file, err := os.Open("test.txt")
if err != nil {
fmt.Printf("打开文件失败: %vn", err)
return
}
defer file.Close()
fileSize := getFileSize(file)
// 并发数，可根据实际需求调整
concurrency := 4
sectionSize := fileSize / int64(concurrency)
resultChan := make(chan string, concurrency)
var wg sync.WaitGroup
// 启动多个goroutine读取不同区间
for i := 0; i < concurrency; i++ {
wg.Add(1)
offset := int64(i) * sectionSize
// 最后一个区间读取剩余所有内容
length := sectionSize
if i == concurrency-1 {
length = fileSize - offset
}
go readSection(file, offset, length, resultChan, &wg)
}
// 等待所有goroutine完成
go func() {
wg.Wait()
close(resultChan)
}()
// 汇总结果
var allContent string
for content := range resultChan {
allContent += content
}
fmt.Printf("读取完成，总内容长度: %dn", len(allContent))
}
并发安全的关键保障
并发读取过程中需要重点注意三个安全问题，避免程序出现异常或数据错误。
1. 文件指针的并发访问问题
同一个os.File实例的文件指针是共享的，多个goroutine同时调用Seek或Read会导致指针位置混乱，因此每个goroutine在开始读取前必须先调用Seek方法将指针移动到自己的区间起始位置，保证各自操作独立的读取范围。
2. 数据竞争问题
多个goroutine不要直接操作同一个共享变量，而是通过channel传递读取结果，由单独的协程汇总数据，避免多个goroutine同时修改同一块内存导致的竞争问题。上述示例中通过resultChan传递每个区间的读取内容，就是典型的无锁安全传递方式。
3. 资源释放问题
使用sync.WaitGroup等待所有读取goroutine完成后再关闭结果通道，避免通道提前关闭导致向通道发送数据时触发panic，同时要及时关闭打开的文件句柄，防止资源泄露。
效率优化建议
要进一步提升并发读取的效率，可以参考以下优化方向：
- 合理设置并发数，一般设置为CPU核心数的1到2倍，过多并发会导致goroutine切换开销增大，反而降低效率。
- 读取缓冲区的大小可以根据文件大小和并发数调整，避免缓冲区过小导致频繁IO操作。
- 如果只需要统计行数、关键词出现次数等不需要完整内容的场景，可以在每个goroutine内部直接完成统计，只传递统计结果，减少通道传递的数据量。
按行数拆分的简化实现
如果文件行数固定且不大，也可以按行数拆分，实现逻辑更简单，不需要处理字节偏移量的不完整行问题。
package main
import (
"bufio"
"fmt"
"os"
"sync"
)
func readByLine(file *os.File, startLine int, endLine int, resultChan chan<- []string, wg *sync.WaitGroup) {
defer wg.Done()
file.Seek(0, 0)
scanner := bufio.NewScanner(file)
currentLine := 0
var lines []string
for scanner.Scan() {
if currentLine >= startLine && currentLine < endLine {
lines = append(lines, scanner.Text())
}
if currentLine >= endLine {
break
}
currentLine++
}
resultChan <- lines
}
func main() {
file, err := os.Open("test.txt")
if err != nil {
fmt.Printf("打开文件失败: %vn", err)
return
}
defer file.Close()
// 先统计总行数，实际场景可以提前获取或预估
totalLines := 1000
concurrency := 4
linesPerGoroutine := totalLines / concurrency
resultChan := make(chan []string, concurrency)
var wg sync.WaitGroup
for i := 0; i < concurrency; i++ {
wg.Add(1)
start := i * linesPerGoroutine
end := start + linesPerGoroutine
if i == concurrency-1 {
end = totalLines
}
go readByLine(file, start, end, resultChan, &wg)
}
go func() {
wg.Wait()
close(resultChan)
}()
var allLines []string
for lines := range resultChan {
allLines = append(allLines, lines...)
}
fmt.Printf("读取总行数: %dn", len(allLines))
}