Golang HTTP Handler：能否在单个流式POST中实现请求响应模式？

核心问题在于HTTP响应默认是缓冲模式：服务器写入的数据会暂存在内存缓冲区中，只有当缓冲区填满、连接关闭或超时后，才会将数据发送给客户端。要实现「发块-收响应-发下块」的循环，必须强制服务器每次写完响应后立刻刷新缓冲区，同时确保请求/响应都采用分块传输编码。

解决方案步骤 #

• 显式启用分块传输：通过响应头Transfer-Encoding: chunked告知客户端这是流式响应，同时不能设置Content-Length（分块传输与固定长度冲突）。
• 强制刷新响应缓冲区：利用Go标准库的http.Flusher接口，每次写完响应后调用Flush()，将缓冲区数据立刻推送给客户端。
• 客户端配合流式读写：不能一次性发送所有数据块，必须发送一块后等待并读取对应响应，再发送下一块。

修改后的服务器代码示例 #

import (
    "bufio"
    "fmt"
    "io"
    "net/http"
    "your-project/logger"
    "your-project/psql"
    "your-project/serverCache"
)

func UserSyncHandler(db *psql.Database, cache *serverCache.Cache) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        // 配置响应头：启用分块传输、禁用缓存
        w.Header().Set("Transfer-Encoding", "chunked")
        w.Header().Set("Cache-Control", "no-cache")

        // 断言响应支持刷新操作
        flusher, ok := w.(http.Flusher)
        if !ok {
            writeError(w, "Response writer does not support flushing", nil, http.StatusInternalServerError)
            return
        }

        reader := bufio.NewReader(r.Body)
        for {
            line, err := reader.ReadBytes('\n')
            if err != nil {
                if err == io.EOF {
                    logger.Info("Received EOF")
                    break
                }
                writeError(w, "Failed to read from sync stream", err, http.StatusInternalServerError)
                return
            }

            // 业务逻辑：处理收到的数据块，生成确认信息
            confirmation := processSyncChunk(line, db, cache)

            // 写入响应并强制刷新
            _, err = w.Write(confirmation)
            if err != nil {
                writeError(w, "Failed to write confirmation", err, http.StatusInternalServerError)
                return
            }
            flusher.Flush()
        }
    }
}

// 业务处理函数：替换为你的数据注册、同步逻辑
func processSyncChunk(data []byte, db *psql.Database, cache *serverCache.Cache) []byte {
    // 示例：处理数据后返回带标识的确认信息
    return []byte(fmt.Sprintf("ACK: processed chunk %s\n", string(data[:len(data)-1])))
}

// 带刷新的错误处理函数
func writeError(w http.ResponseWriter, msg string, err error, statusCode int) {
    w.WriteHeader(statusCode)
    errMsg := fmt.Sprintf("%s", msg)
    if err != nil {
        errMsg = fmt.Sprintf("%s: %v", msg, err)
    }
    _, _ = w.Write([]byte(errMsg + "\n"))
    if flusher, ok := w.(http.Flusher); ok {
        flusher.Flush()
    }
}

客户端配合要点 #

客户端必须采用流式写入请求体，且每发送一块数据后，立刻读取服务器返回的对应响应，再发送下一块。以下是Go客户端示例片段：

func syncClient(url string, dataChunks [][]byte) error {
    req, err := http.NewRequest("POST", url, nil)
    if err != nil {
        return err
    }
    req.Body = &chunkedRequestBody{chunks: dataChunks}
    req.Header.Set("Transfer-Encoding", "chunked")

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    respReader := bufio.NewReader(resp.Body)
    for idx := range dataChunks {
        // 读取对应块的响应
        respLine, err := respReader.ReadBytes('\n')
        if err != nil {
            return fmt.Errorf("read chunk %d response failed: %w", idx, err)
        }
        // 处理响应（比如验证确认信息）
        fmt.Printf("Chunk %d confirmed: %s", idx, string(respLine))
    }
    return nil
}

// 自定义RequestBody实现流式分块发送
type chunkedRequestBody struct {
    chunks [][]byte
    currentIdx int
    currentPos int
}

func (cr *chunkedRequestBody) Read(p []byte) (n int, err error) {
    if cr.currentIdx >= len(cr.chunks) {
        return 0, io.EOF
    }
    currentChunk := cr.chunks[cr.currentIdx][cr.currentPos:]
    n = copy(p, currentChunk)
    cr.currentPos += n
    if cr.currentPos == len(cr.chunks[cr.currentIdx]) {
        cr.currentIdx++
        cr.currentPos = 0
    }
    return n, nil
}

额外注意事项 #

• 必须使用HTTP/1.1或更高版本：HTTP/1.0不支持分块传输编码。
• 避免中间代理缓存：设置Cache-Control: no-cache防止代理缓存响应块，导致客户端无法即时收到数据。
• 错误处理要及时：一旦读写出错，立即关闭连接，避免客户端长时间等待。

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