refactor(uploader): 优化文件上传并发处理逻辑
构建上传工具 / build-tool (push) Successful in 1m46s

- 将初始化连接池的并发方式改为waitgroup控制的goroutine池
- 调整文件处理时的channel缓冲区大小从100增加到200
- 移除不必要的sync.WaitGroup变量声明
- 修改进度计算逻辑,确保上传完成时进度显示为100%
- 添加对processLines函数的功能注释
- 优化上下文取消时的资源清理流程,及时关闭channel
This commit is contained in:
2026-04-28 15:43:02 +08:00
parent 1cac9e9013
commit 73a7d26816
2 changed files with 42 additions and 23 deletions
+23 -13
View File
@@ -211,7 +211,6 @@ func getTxtFiles(dir string) (txtFiles []string, err error) {
}
func processFile(ctx context.Context, logChan *chan string, filePath string, fileLines int) {
var wg sync.WaitGroup
// 打开文件
file, err := os.Open(filePath)
if err != nil {
@@ -221,8 +220,7 @@ func processFile(ctx context.Context, logChan *chan string, filePath string, fil
defer file.Close()
// 创建行通道
lines := make(chan string, 100)
defer close(lines)
lines := make(chan string, 200)
var countLine int32 = 0
// 创建指定个worker同时处理文件上传
for i := 0; i < config.APPConfig.ThreadCount; i++ {
@@ -230,9 +228,9 @@ func processFile(ctx context.Context, logChan *chan string, filePath string, fil
case <-ctx.Done():
return
default:
wg.Go(func() {
go func() {
processLines(ctx, logChan, &lines, i, filePath, &countLine)
})
}()
}
}
@@ -255,22 +253,33 @@ func processFile(ctx context.Context, logChan *chan string, filePath string, fil
}
}()
// 等待所有行处理完成并推送进度
for int(countLine) != fileLines {
select {
case <-ctx.Done():
wg.Wait()
close(lines) //关闭processLines中的上传线程
return
default:
progress.Store(filepath.Base(filePath),
Progress{FileName: filepath.Base(filePath),
Total: fileLines, Uploaded: int(countLine),
Percentage: int(float64(countLine)/float64(fileLines)*100) + 1,
})
time.Sleep(500 * time.Millisecond)
Progress{
FileName: filepath.Base(filePath),
Total: fileLines,
Uploaded: int(countLine),
Percentage: int(float64(countLine) / float64(fileLines) * 100),
},
)
}
}
wg.Wait()
//上传完成,进度设为100
progress.Store(filepath.Base(filePath),
Progress{
FileName: filepath.Base(filePath),
Total: fileLines,
Uploaded: int(countLine),
Percentage: 100,
},
)
close(lines) //关闭processLines中的上传线程
if err := scanner.Err(); err != nil {
AddLog(logChan, fmt.Sprintf("读取文件 %s 错误: %v", filePath, err))
@@ -280,6 +289,7 @@ func processFile(ctx context.Context, logChan *chan string, filePath string, fil
AddLog(logChan, fmt.Sprintf("文件【%s】处理完成,共处理 %d 行数据", filepath.Base(filePath), countLine))
}
// processLines 处理接受到的每一行数据并上传(chan 管道接受数据)
func processLines(ctx context.Context, logChan *chan string, lines *chan string, workerID int, filePath string, countLine *int32) {
for line := range *lines {
select {