wangjia
3c687e5e0b
新包 internal/ratelimit：内存/redis(GCRA+Lua) 双实现 + 出错逐调用降级内存 （fail-open 到内存不 fail-closed）。REDIS_ADDR 空=内存模式，行为与既往一致； 配置后跨重启保状态、支持多实例。miniredis 全覆盖测试，零真实外部依赖。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
107 lines
2.5 KiB
Go
107 lines
2.5 KiB
Go
package ratelimit
|
||
|
||
import (
|
||
"log"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"golang.org/x/time/rate"
|
||
)
|
||
|
||
// fallbackStore：redis 为主、内存为影子。每次调用先试 redis，出错即降级到
|
||
// 对应的内存实现（fail-open 到内存而非 fail-closed——redis 挂掉时防护降级但
|
||
// 不中断，业务请求不 5xx）。redis 恢复后自动回主路径。降级日志限频打印。
|
||
|
||
type fallbackStore struct {
|
||
redis *redisStore
|
||
mem *memoryStore
|
||
}
|
||
|
||
func newFallbackStore(rdb *redis.Client) *fallbackStore {
|
||
return &fallbackStore{redis: newRedisStore(rdb), mem: newMemoryStore()}
|
||
}
|
||
|
||
// warnDegraded 限频告警（每 30s 至多一条，防 redis 宕机刷爆日志）。
|
||
var (
|
||
warnMu sync.Mutex
|
||
warnLast time.Time
|
||
)
|
||
|
||
func warnDegraded(err error) {
|
||
warnMu.Lock()
|
||
defer warnMu.Unlock()
|
||
if time.Since(warnLast) < 30*time.Second {
|
||
return
|
||
}
|
||
warnLast = time.Now()
|
||
log.Printf("[ratelimit] redis 不可用，已降级内存限流：%v", err)
|
||
}
|
||
|
||
func (s *fallbackStore) NewLimiter(r rate.Limit, burst int) Limiter {
|
||
return &fallbackLimiter{r: s.redis.newLimiter(r, burst), m: s.mem.NewLimiter(r, burst)}
|
||
}
|
||
func (s *fallbackStore) Counter() Counter {
|
||
return &fallbackCounter{r: &redisCounter{rdb: s.redis.rdb}, m: s.mem.Counter()}
|
||
}
|
||
func (s *fallbackStore) FailLocker() FailLocker {
|
||
return &fallbackFailLocker{r: &redisFailLocker{rdb: s.redis.rdb}, m: s.mem.FailLocker()}
|
||
}
|
||
|
||
type fallbackLimiter struct {
|
||
r *redisLimiter
|
||
m Limiter
|
||
}
|
||
|
||
func (l *fallbackLimiter) Allow(key string) Result {
|
||
res, err := l.r.allow(key)
|
||
if err != nil {
|
||
warnDegraded(err)
|
||
return l.m.Allow(key)
|
||
}
|
||
return res
|
||
}
|
||
|
||
type fallbackCounter struct {
|
||
r *redisCounter
|
||
m Counter
|
||
}
|
||
|
||
func (c *fallbackCounter) Incr(key string, ttl time.Duration) (int64, error) {
|
||
n, err := c.r.Incr(key, ttl)
|
||
if err != nil {
|
||
warnDegraded(err)
|
||
return c.m.Incr(key, ttl)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
type fallbackFailLocker struct {
|
||
r *redisFailLocker
|
||
m FailLocker
|
||
}
|
||
|
||
func (l *fallbackFailLocker) Locked(key string) bool {
|
||
locked, err := l.r.locked(key)
|
||
if err != nil {
|
||
warnDegraded(err)
|
||
return l.m.Locked(key)
|
||
}
|
||
return locked
|
||
}
|
||
|
||
func (l *fallbackFailLocker) RecordFailure(key string, max int, lockFor time.Duration) {
|
||
if err := l.r.recordFailure(key, max, lockFor); err != nil {
|
||
warnDegraded(err)
|
||
l.m.RecordFailure(key, max, lockFor)
|
||
}
|
||
}
|
||
|
||
func (l *fallbackFailLocker) Reset(key string) {
|
||
// 双清：降级期间可能在内存里积了计数，恢复后一并清掉。
|
||
if err := l.r.reset(key); err != nil {
|
||
warnDegraded(err)
|
||
}
|
||
l.m.Reset(key)
|
||
}
|