Golang双通道收发模式实现：API响应异步写入文件方案咨询

你的代码核心问题在于单次select只能处理一个响应或触发一次超时，没法持续监听respChan处理所有API返回结果，同时通道关闭时机、goroutine等待逻辑也有疏漏。以下是串联完整数据流的实现方案：

关键问题梳理 #

• 原代码里的select仅执行一次，无法处理所有并发请求返回的响应
• respChan的defer close放在main开头，会在main结束前提前关闭，此时可能还有请求goroutine在往通道发数据，直接导致panic
• 没有等待所有写入文件的goroutine完成，可能出现程序退出时文件还没写完的情况
• 超时逻辑应该针对整个接收周期，而非单次响应

完整实现代码 #

package main

import (
    "fmt"
    "sync"
    "time"
)

// 假设的Request和Response结构体，根据实际定义调整
type Request struct {
    // 请求相关字段
}

type Response struct {
    // API响应元数据字段
}

func (r Request) Get() (Response, error) {
    // 模拟API调用延迟
    time.Sleep(100 * time.Millisecond)
    return Response{}, nil
}

func Write(resp Response) (string, error) {
    // 模拟写入文件逻辑，返回文件路径
    time.Sleep(50 * time.Millisecond)
    return "/tmp/test.txt", nil
}

func main() {
    var reqWg sync.WaitGroup
    var writeWg sync.WaitGroup

    // 1. 初始化通道：respChan用于传递API响应，writeChan传递写入后的文件路径
    respChan := make(chan Response)
    writeChan := make(chan string)

    // 模拟请求列表
    requests := []Request{{}, {}, {}, {}}

    // 2. 启动并发API请求
    for _, req := range requests {
        reqWg.Add(1)
        go func(r Request) {
            defer reqWg.Done()
            resp, err := r.Get()
            if err != nil {
                fmt.Printf("API请求失败: %v\n", err)
                return
            }
            respChan <- resp
        }(req)
    }

    // 3. 单独goroutine：等待所有请求完成后关闭respChan，触发后续for range退出
    go func() {
        reqWg.Wait()
        close(respChan)
        fmt.Println("所有API请求已完成，关闭respChan")
    }()

    // 4. 持续监听respChan，为每个响应启动写入goroutine
    go func() {
        for resp := range respChan {
            writeWg.Add(1)
            go func(response Response) {
                defer writeWg.Done()
                filePath, err := Write(response)
                if err != nil {
                    fmt.Printf("写入文件失败: %v\n", err)
                    return
                }
                writeChan <- filePath
            }(resp)
        }
        // 所有写入goroutine完成后关闭writeChan
        writeWg.Wait()
        close(writeChan)
        fmt.Println("所有文件写入完成，关闭writeChan")
    }()

    // 5. 处理超时逻辑：如果15秒内没有任何响应或处理完成，退出程序
    timeout := time.After(15 * time.Second)

    // 6. 持续监听writeChan，处理后续逻辑（比如记录文件路径、做后续处理）
    for {
        select {
        case filePath, ok := <-writeChan:
            if !ok {
                // writeChan已关闭，所有处理完成
                fmt.Println("所有文件路径已接收完成")
                return
            }
            fmt.Printf("文件已写入: %s\n", filePath)
            // 这里可以添加后续处理逻辑，比如上传、索引等
        case <-timeout:
            fmt.Println("15秒无响应，程序退出")
            return
        }
    }
}

核心逻辑说明 #

• 通道关闭时机：用单独goroutine等待请求WaitGroup完成后关闭respChan，确保所有响应都能被发送到通道
• 持续接收响应：用for range respChan替代单次select，自动持续接收直到通道关闭
• 写入goroutine管理：新增writeWg等待所有写入操作完成，避免程序提前退出导致文件写入不完整
• 超时全局监听：将超时逻辑放在最终的select里，监听整个处理流程的超时情况
• 数据流串联：API请求 → respChan → 写入goroutine → writeChan → 后续处理，形成完整的异步数据流

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