Golang并发处理文件读写与网络请求

sync.WaitGroup

go

go func() {
    // 执行文件读取或网络请求
    // ...
}()

results := make(chan string, numTasks) // 存储结果
errors := make(chan error, numTasks)   // 存储错误

sync.WaitGroup

sync.WaitGroup

package main

import (
    "context"
    "fmt"
    "io"
    "net/http"
    "os"
    "path/filepath"
    "sync"
    "time"
)

// downloadAndSaveFunc 负责下载一个URL并保存到文件
func downloadAndSaveFunc(ctx context.Context, url string, filename string, wg *sync.WaitGroup, results chan<- string, errs chan<- error) {
    defer wg.Done()

    select {
    case <-ctx.Done():
        errs <- fmt.Errorf("下载 %s 已取消: %w", url, ctx.Err())
        return
    default:
        // 继续执行
    }

    fmt.Printf("开始下载: %s -> %s\n", url, filename)

    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        errs <- fmt.Errorf("创建请求失败 %s: %w", url, err)
        return
    }

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        errs <- fmt.Errorf("下载 %s 失败: %w", url, err)
        return
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        errs <- fmt.Errorf("下载 %s 失败，状态码: %d", url, resp.StatusCode)
        return
    }

    file, err := os.Create(filename)
    if err != nil {
        errs <- fmt.Errorf("创建文件 %s 失败: %w", filename, err)
        return
    }
    defer file.Close()

    _, err = io.Copy(file, resp.Body)
    if err != nil {
        errs <- fmt.Errorf("保存文件 %s 失败: %w", filename, err)
        return
    }

    results <- fmt.Sprintf("成功下载并保存: %s -> %s", url, filename)
}

func main() {
    urls := []string{
        "https://www.google.com/robots.txt",
        "https://www.github.com/robots.txt",
        "https://www.bing.com/robots.txt",
        "https://www.golang.org/robots.txt", // 可能会因网络问题访问不到
        "https://httpbin.org/delay/5", // 模拟一个慢请求
    }

    outputDir := "downloads"
    os.MkdirAll(outputDir, os.ModePerm) // 确保输出目录存在

    var wg sync.WaitGroup
    results := make(chan string, len(urls))
    errs := make(chan error, len(urls))

    // 设置一个带超时的主Context
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel() // 确保所有操作完成后取消Context

    for i, url := range urls {
        filename := filepath.Join(outputDir, fmt.Sprintf("content_%d.txt", i+1))
        wg.Add(1)
        go downloadAndSaveFunc(ctx, url, filename, &wg, results, errs)
    }

    // 启动一个goroutine来等待所有任务完成，并在完成后关闭results和errs channel
    go func() {
        wg.Wait()
        close(results)
        close(errs)
    }()

    // 收集结果和错误
    var successfulDownloads []string
    var downloadErrors []error

    // 使用select非阻塞地从results和errs channel中读取
    for {
        select {
        case res, ok := <-results:
            if !ok { // channel已关闭
                results = nil // 防止再次从已关闭的channel读取
                break
            }
            successfulDownloads = append(successfulDownloads, res)
        case err, ok := <-errs:
            if !ok { // channel已关闭
                errs = nil
                break
            }
            downloadErrors = append(downloadErrors, err)
        case <-time.After(100 * time.Millisecond): // 短暂等待，防止CPU空转
            if results == nil && errs == nil { // 所有channel都已处理完毕
                goto END_COLLECTION
            }
        }
        if results == nil && errs == nil { // 再次检查，确保在跳出循环前捕获所有
            break
        }
    }
END_COLLECTION:

    fmt.Println("\n--- 下载结果 ---")
    for _, res := range successfulDownloads {
        fmt.Println(res)
    }
    fmt.Println("\n--- 错误报告 ---")
    if len(downloadErrors) == 0 {
        fmt.Println("无错误。")
    } else {
        for _, err := range downloadErrors {
            fmt.Printf("错误: %v\n", err)
        }
    }
}

WaitGroup

context.Context

context.Context

http.Request

Context()

WithContext()

package main

import (
    "context"
    "fmt"
    "io"
    "net/http"
    "time"
)

func fetchURLWithContext(ctx context.Context, url string) (string, error) {
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil) // 将Context绑定到请求
    if err != nil {
        return "", fmt.Errorf("创建请求失败: %w", err)
    }

    client := &http.Client{}
    resp, err := client.Do(req) // 发起请求
    if err != nil {
        // 这里会捕获到context.DeadlineExceeded或context.Canceled错误
        return "", fmt.Errorf("网络请求失败: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return "", fmt.Errorf("请求返回非OK状态: %d", resp.StatusCode)
    }

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return "", fmt.Errorf("读取响应体失败: %w", err)
    }
    return string(body), nil
}

func main() {
    // 场景一：设置一个5秒的超时
    ctxTimeout, cancelTimeout := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancelTimeout() // 最佳实践：确保Context最终被取消

    fmt.Println("--- 尝试一个可能超时的请求 ---")
    // 这是一个模拟的慢请求