如何优雅关闭SSE连接并读取完整响应体？为何加Sleep后失效？

我正在测试一个采用Server-Sent Events（SSE）返回JSON流的HTTP端点，取消请求的context.Context后，json.Decode(resp.Body)返回‘context cancelled’错误。请问如何优雅地完成请求以读取所有已接收的响应体？是否需要在服务器端进行相关处理？

更新：不稳定的复现示例 #

我制作了最小可复现示例，发现可以读取已取消请求的响应体，但示例存在不稳定问题——在cancel后添加time.Sleep就无法读取响应体，这是为什么？

示例代码 #

package main

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

func Subscribe(w http.ResponseWriter, req *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming unsupported", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("Access-Control-Allow-Origin", "*")
    flusher.Flush()

    tick := time.Tick(500 * time.Millisecond)
    for {
        select {
        case <-req.Context().Done():
            return
        case <-tick:
            fmt.Fprintln(w, "EVENT!")
            flusher.Flush()
        }
    }
}

func main() {
    s := httptest.NewServer(http.HandlerFunc(Subscribe))
    defer s.Close()
    c := s.Client()

    var sub *http.Response
    var cancelSub context.CancelFunc
    {
        ctx := context.Background()
        ctx, cancelSub = context.WithCancel(ctx)
        req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.URL+"/subscribe", nil)
        if err != nil {
            panic(err.Error())
        }
        sub, err = c.Do(req)
        if err != nil {
            panic(err.Error())
        }
        defer sub.Body.Close()
        if sub.StatusCode != 200 {
            panic("StatusCode != 200")
        }
    }

    time.Sleep(3 * time.Second)
    cancelSub()

    // WITH THIS SLEEP NO BODY IS PRINTED?!
    time.Sleep(time.Second)

    b, err := io.ReadAll(sub.Body)
    fmt.Printf("error: %v\nbody:\n%s\n", err, b)
}

输出对比 #

• 添加time.Sleep(time.Second)后的输出：

error: context canceled
body:

• 未添加该Sleep的输出：

error: context canceled
body:
EVENT!
EVENT!
EVENT!
EVENT!
EVENT!

1. 为什么cancel后加Sleep就无法读取响应体？ #

调用cancelSub()取消请求上下文时，客户端的http.Transport会立即触发连接关闭流程。如果cancel后立刻调用io.ReadAll，内核缓冲区中可能还留存着服务器已发送但未被用户态读取的响应数据，因此能读到内容。

但添加time.Sleep(time.Second)后，客户端传输层会在这段时间内完成连接关闭与资源清理，内核缓冲区里的未读数据会被丢弃，后续读取resp.Body就会直接返回context canceled错误，且读不到任何内容。

另外，服务器端收到上下文取消信号后直接return，未主动关闭响应写入流，会导致服务器端连接清理存在延迟，但客户端的连接关闭优先级更高，最终还是会先切断连接。

2. 如何优雅读取所有已接收的响应体？ #

要确保读取到所有已接收的响应数据，核心是在取消上下文之前启动异步读取，或者取消后立刻读取，避免给连接关闭留时间窗口，推荐以下方案：

• 异步提前读取（可靠方案）：发送请求后，启动goroutine持续读取resp.Body并缓存到缓冲区（如bytes.Buffer），直到读取结束。即使上下文被取消，也能从缓冲区获取所有已接收的数据。
  示例代码片段：

  var buf bytes.Buffer
  done := make(chan error)
  go func() {
      _, err := io.Copy(&buf, sub.Body)
      done <- err
  }()
  
  // 等待一段时间后取消请求
  time.Sleep(3 * time.Second)
  cancelSub()
  
  // 等待读取完成
  err := <-done
  // buf中留存所有已接收的响应数据
  fmt.Printf("error: %v\nbody:\n%s\n", err, buf.String())
  

• 取消后立即读取（临时方案）：去掉cancel后的Sleep，直接调用读取方法，抢在连接关闭前读取内核缓冲区数据。但这种方式依赖时机，可靠性低，仅适合临时调试。

3. 服务器端是否需要处理？ #

需要做优化，让连接关闭流程更可控：

• 收到上下文取消信号时，不要直接return，可以先发送一个SSE结束事件（如event: close），调用flusher.Flush()确保事件发送，再优雅退出。虽然客户端可能因连接关闭读不到最后这个事件，但能规范服务器端资源清理。
• 避免资源泄漏，确保响应写入流的相关资源被正确释放。

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