如何在HTTP响应返回客户端后执行GCP Pub/Sub推送并避免性能损耗

你的中间件性能问题根源在于同步阻塞+全量内存缓存响应：
当前逻辑是先让业务handler把完整响应写入内存Recorder，再将Recorder内容复制给客户端，最后同步执行Pub/Sub推送——这导致客户端必须等待handler处理、内存复制、Pub/Sub推送全部完成才能结束请求，同时内存复制也额外增加了开销。

要实现「不影响API延迟」的推送，核心是让响应实时返回给客户端，同时异步执行Pub/Sub操作，并且避免全量缓存响应的额外开销。

1. 自定义ResponseWriter实现流式捕获响应 #

首先实现一个自定义的ResponseWriter，在向客户端写入响应的同时，同步记录响应状态码、Header和响应体，不需要先全量缓存到内存：

import (
    "bytes"
    "net/http"
    "strconv"
    "log"
    "cloud.google.com/go/pubsub"
)

// 自定义ResponseWriter，同时向客户端写入并记录响应内容
type responseRecorder struct {
    http.ResponseWriter
    statusCode int
    body       *bytes.Buffer
}

func newResponseRecorder(w http.ResponseWriter) *responseRecorder {
    return &responseRecorder{
        ResponseWriter: w,
        statusCode:     http.StatusOK,
        body:           &bytes.Buffer{},
    }
}

// 重写WriteHeader，记录状态码
func (rr *responseRecorder) WriteHeader(code int) {
    rr.statusCode = code
    rr.ResponseWriter.WriteHeader(code)
}

// 重写Write，同时向客户端写入和记录响应体
func (rr *responseRecorder) Write(b []byte) (int, error) {
    // 先向客户端写入响应
    n, err := rr.ResponseWriter.Write(b)
    if err != nil {
        return n, err
    }
    // 同步记录响应体到内存
    rr.body.Write(b[:n])
    return n, nil
}

2. 修改中间件为异步推送 #

在中间件中使用自定义responseRecorder，让业务handler直接向客户端写入响应，然后启动goroutine异步执行Pub/Sub推送，完全不阻塞请求流程：

// 全局初始化Pub/Sub客户端（避免每次请求创建新客户端）
var pubsubClient *pubsub.Client

// 可选：控制Pub/Sub推送的并发数，避免goroutine爆炸
var pubsubWorkerSemaphore = make(chan struct{}, 100)

func MyPush(handler http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 创建自定义Recorder，流式捕获响应
        rr := newResponseRecorder(w)
        
        // 让业务handler直接向客户端写入响应，无额外内存复制开销
        handler.ServeHTTP(rr, r)
        
        // 异步执行Pub/Sub推送，不阻塞当前请求
        go func(ctx context.Context, status int, body []byte, path, method string) {
            // 使用信号量控制并发数
            pubsubWorkerSemaphore <- struct{}{}
            defer func() { <-pubsubWorkerSemaphore }()
            
            // 构造Pub/Sub消息
            msg := &pubsub.Message{
                Data: body,
                Attributes: map[string]string{
                    "status_code": strconv.Itoa(status),
                    "request_path": path,
                    "request_method": method,
                },
            }
            
            // 推送消息到指定Topic
            _, err := pubsubClient.Topic("your-topic-id").Publish(ctx, msg).Get(ctx)
            if err != nil {
                log.Printf("Pub/Sub推送失败: 请求路径=%s, 错误=%v", path, err)
            }
        }(r.Context(), rr.statusCode, rr.body.Bytes(), r.URL.Path, r.Method)
    })
}

关键注意事项 #

• 上下文传递：用r.Context()关联请求生命周期，如果客户端中途断开连接，Pub/Sub推送可以自动取消，避免无效操作。
• 并发控制：用信号量限制Pub/Sub推送的并发数，防止高流量下goroutine耗尽系统资源。
• 大响应处理：如果响应体超过Pub/Sub单条消息上限（10MB），可以先将响应存入GCS，再把GCS链接推送到Pub/Sub。
• 可靠性保障：如果需要100%不丢消息，异步推送可能存在进程崩溃丢失消息的风险，此时可以先将消息写入本地持久化队列（如Redis），再用后台消费者异步推送到Pub/Sub。

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