如何编写中间件优化PostgreSQL，应对每秒5万消息写入瓶颈？

完全可以通过中间件+配套策略优化这个场景，核心思路是削峰填谷，把瞬时超过数据库处理能力的请求缓存后匀速写入，同时保证服务稳定性。以下是针对Go语言服务的具体落地方案：

1. 令牌桶限流中间件 #

用Go官方生态的golang.org/x/time/rate实现入口限流，直接匹配数据库处理能力拦截超额请求：

• 逻辑：在HTTP请求入口处，用令牌桶算法控制每秒仅放行30000个请求（与数据库处理能力对齐），超额请求要么返回重试信号，要么放入等待队列缓冲
• 简化代码示例：

import (
    "net/http"
    "golang.org/x/time/rate"
)

func rateLimitMiddleware(next http.Handler) http.Handler {
    // 每秒生成3万令牌，桶容量设为5万应对瞬时峰值
    limiter := rate.NewLimiter(rate.Limit(30000), 50000)
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if !limiter.Allow() {
            w.WriteHeader(http.StatusServiceUnavailable)
            w.Write([]byte("请求过多，请稍后重试"))
            return
        }
        next.ServeHTTP(w, r)
    })
}

// 注册中间件到消息摄入接口
http.Handle("/ingest", rateLimitMiddleware(ingestHandler))

2. 异步批量写入中间件 #

在HTTP接收层与数据库层之间加异步缓冲层，将请求先存入带缓冲的channel，再批量写入数据库：

• 逻辑：HTTP handler接收消息后直接发送到channel并返回成功，后台goroutine从channel批量拉取消息（比如每次100条），用PostgreSQL的COPY命令或批量INSERT写入，控制每秒写入总量不超3万
• 核心代码示例：

import (
    "io"
    "net/http"
    "database/sql"
    "time"
    _ "github.com/lib/pq"
)

// 缓冲channel，容量设为10万应对峰值
var msgChan = make(chan []byte, 100000)

func init() {
    // 启动3个消费goroutine，根据数据库处理能力调整数量
    for i := 0; i < 3; i++ {
        go batchWriteToDB()
    }
}

func asyncIngestMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        msg, _ := io.ReadAll(r.Body)
        select {
        case msgChan <- msg:
            w.WriteHeader(http.StatusAccepted)
            w.Write([]byte("消息已接收"))
        default:
            w.WriteHeader(http.StatusServiceUnavailable)
            w.Write([]byte("系统繁忙，请稍后重试"))
        }
    })
}

func batchWriteToDB() {
    db, _ := sql.Open("postgres", "your_dsn_here")
    defer db.Close()

    batch := make([][]byte, 0, 100)
    // 每10ms触发一次批量写入，避免等待过久
    ticker := time.NewTicker(10 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case msg := <-msgChan:
            batch = append(batch, msg)
            if len(batch) >= 100 {
                flushBatch(db, batch)
                batch = batch[:0]
            }
        case <-ticker.C:
            if len(batch) > 0 {
                flushBatch(db, batch)
                batch = batch[:0]
            }
        }
    }
}

func flushBatch(db *sql.DB, batch [][]byte) {
    // 构造批量INSERT语句或使用PostgreSQL COPY命令提升写入效率
    // 此处省略具体SQL拼接与执行逻辑
}

3. 本地缓存降级中间件 #

如果峰值持续超出数据库能力，可引入本地内存缓存（如bigcache）作为临时存储：

• 逻辑：当数据库写入压力过高时，将消息先写入本地缓存，后台goroutine定时将缓存数据同步到数据库
• 优势：比纯内存channel更稳定，支持自动淘汰、过期时间等特性，避免内存溢出
• 注意：需添加服务重启时的缓存持久化恢复逻辑，保证消息不丢失

关键注意事项 #

• 消息可靠性：若业务要求消息零丢失，需给异步队列加持久化兜底（如内存队列满时写入本地文件），或替换为Redis、Kafka等持久化队列
• 监控告警：在中间件中加入队列长度、每秒接收/写入量等监控指标，队列超阈值时触发告警
• 客户端重试：返回503时需明确告知客户端重试策略（如指数退避），避免频繁重试加剧服务压力

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