refactor(uploader): 优化文件上传并发处理逻辑

- 将初始化连接池的并发方式改为waitgroup控制的goroutine池
- 调整文件处理时的channel缓冲区大小从100增加到200
- 移除不必要的sync.WaitGroup变量声明
- 修改进度计算逻辑，确保上传完成时进度显示为100%
- 添加对processLines函数的功能注释
- 优化上下文取消时的资源清理流程，及时关闭channel

internal/api/api.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"sync"
 	"time"
 )
 
@@ -28,7 +29,11 @@ func init() {
 
 // InitConn 创建连接池
 func InitConn() {
-	for i := 0; i < 200; i++ {
+	wg := &sync.WaitGroup{}
+
+	for i := 0; i < 10; i++ {
+		wg.Go(func() {
+			for i := 0; i < 50; i++ {
 				resp, err := httpClient.Get(config.APPConfig.Url + "/api/test")
 				if err != nil {
 					fmt.Println(err)
@@ -39,6 +44,10 @@ func InitConn() {
 					resp.Body.Close()
 				}()
 			}
+		})
+	}
+
+	wg.Wait()
 }
 
 func UploadDataToServer(ctx context.Context, data string) error {

internal/uploader/uploader.go

@@ -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 {