Go语言中实现HTTP响应流式传输的策略与实践

package main

import (
    "fmt"
    "log"
    "net/http"
    "time"
)

func handleStream(res http.ResponseWriter, req *http.Request) {
    // 设置Content-Type为text/plain或text/event-stream等，
    // 告知客户端这是一个流式响应。
    res.Header().Set("Content-Type", "text/plain; charset=utf-8")
    res.Header().Set("X-Content-Type-Options", "nosniff") // 防止浏览器嗅探类型

    fmt.Fprintf(res, "发送第一行数据...\n")
    // 尝试刷新缓冲区
    if f, ok := res.(http.Flusher); ok {
        f.Flush()
    } else {
        log.Println("警告: http.ResponseWriter 未实现 http.Flusher 接口。")
    }

    time.Sleep(2 * time.Second) // 模拟耗时操作

    fmt.Fprintf(res, "发送第二行数据...\n")
    if f, ok := res.(http.Flusher); ok {
        f.Flush()
    }

    time.Sleep(2 * time.Second) // 模拟耗时操作

    fmt.Fprintf(res, "发送第三行数据并结束。\n")
    // 最后一次写入后通常不需要显式Flush，因为请求结束时会自动处理。
    // 但如果希望立即结束流，可以再次Flush。
    if f, ok := res.(http.Flusher); ok {
        f.Flush()
    }
}

func main() {
    http.HandleFunc("/stream", handleStream)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
    "fmt"
    "io"
    "log"
    "net/http"
    "os/exec"
    "time"
)

const BUF_LEN = 4096 // 缓冲区大小

// writeCmdOutput 协程：从管道读取命令输出并写入HTTP响应
func writeCmdOutput(res http.ResponseWriter, pipeReader *io.PipeReader) {
    defer pipeReader.Close() // 确保管道读取端关闭

    buffer := make([]byte, BUF_LEN)
    for {
        n, err := pipeReader.Read(buffer) // 从管道读取数据
        if err != nil {
            if err == io.EOF {
                // 命令执行完毕，管道写入端已关闭
                break
            }
            log.Printf("读取管道错误: %v", err)
            break
        }

        data := buffer[0:n]
        _, writeErr := res.Write(data) // 将数据写入HTTP响应
        if writeErr != nil {
            log.Printf("写入HTTP响应错误: %v", writeErr)
            break // 客户端可能已断开连接
        }

        // 如果ResponseWriter支持Flusher，则刷新
        if f, ok := res.(http.Flusher); ok {
            f.Flush()
        }

        // 重置缓冲区，避免脏数据（虽然Read通常会覆盖，但显式清零更安全）
        // for i := 0; i < n; i++ {
        //  buffer[i] = 0
        // }
    }
    log.Println("命令输出流式传输协程结束。")
}

func handleExecStream(res http.ResponseWriter, req *http.Request) {
    res.Header().Set("Content-Type", "text/plain; charset=utf-8")
    res.Header().Set("X-Content-Type-Options", "nosniff")

    // 模拟一个长时间运行的命令
    cmd := exec.Command("bash", "-c", "for i in $(seq 1 5); do echo \"Line $i from command...\"; sleep 1; done; echo \"Command finished.\"")

    // 创建管道
    pipeReader, pipeWriter := io.Pipe()
    cmd.Stdout = pipeWriter // 命令的标准输出连接到管道写入端
    cmd.Stderr = pipeWriter // 命令的标准错误输出也连接到管道写入端

    // 启动一个Goroutine来处理管道读取和HTTP响应写入
    go writeCmdOutput(res, pipeReader)

    // 启动命令
    err := cmd.Start()
    if err != nil {
        log.Printf("启动命令失败: %v", err)
        http.Error(res, "无法启动命令", http.StatusInternalServerError)
        // 确保管道写入端关闭，避免读取Goroutine阻塞
        pipeWriter.Close()
        return
    }

    // 等待命令执行完毕
    cmdErr := cmd.Wait()
    if cmdErr != nil {
        log.Printf("命令执行失败: %v", cmdErr)
        // 注意：如果命令在写入管道时失败，错误会通过管道传递给pipeReader
        // 并在writeCmdOutput中处理。这里主要是处理命令启动失败或非零退出码。
    }

    // 命令执行完毕，关闭管道写入端，通知读取Goroutine结束
    pipeWriter.Close()
    log.Println("主Goroutine：命令执行完毕，管道写入端已关闭。")
}

func main() {
    http.HandleFunc("/exec-stream", handleExecStream)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
    "fmt"
    "io"
    "log"
    "net/http"
    "os/exec"
    "time"
)

// 定义一个接口，要求同时实现 http.ResponseWriter 和 http.Flusher
type flusher interface {
    http.ResponseWriter
    http.Flusher
}

// flushWriter 结构体，包装一个 flusher 接口
type flushWriter struct {
    f flusher
}

// Write 方法实现了 io.Writer 接口，并在写入后自动调用 Flush
func (fw flushWriter) Write(p []byte) (int, error) {
    n, err := fw.f.Write(p)
    if err != nil {
        return n, err
    }
    fw.f.Flush() // 每次写入后立即刷新
    return n, nil
}

func handleExecStreamV2(res http.ResponseWriter, req *http.Request) {
    res.Header().Set("Content-Type", "text/plain; charset=utf-8")
    res.Header().Set("X-Content-Type-Options", "nosniff")

    // 检查 res 是否同时实现了 http.ResponseWriter 和 http.Flusher
    flusherRes, ok := res.(flusher)
    if !ok {
        log.Println("警告: http.ResponseWriter 未实现 http.Flusher 接口，无法使用 flushWriter。")
        http.Error(res, "服务器不支持流式传输", http.StatusInternalServerError)
        return
    }

    // 模拟一个长时间运行的命令
    cmd := exec.Command("bash", "-c", "for i in $(seq 1 5); do echo \"Line $i from command (V2)...\"; sleep 1; done; echo \"Command V2 finished.\"")

    // 创建 flushWriter 实例
    fw := flushWriter{f: flusherRes}

    // 将命令的输出直接重定向到 flushWriter
    cmd.Stdout = fw
    cmd.Stderr = fw

    // 启动并等待命令执行
    err := cmd.Run()
    if err != nil {
        log.Printf("命令执行失败: %v", err)
        // 这里不需要额外的关闭操作，因为错误会通过 fw.Write 传递
        // 并且 Flush 也会被调用
    }
    log.Println("主Goroutine：命令 V2 执行完毕。")
}

func main() {
    http.HandleFunc("/stream", handleStream)
    http.HandleFunc("/exec-stream", handleExecStream)
    http.HandleFunc("/exec-stream-v2", handleExecStreamV2) // 新增 V2 路由
    log.Fatal(http.ListenAndServe(":8080", nil))
}