Victor_Jay
5a8b0ca79c
- 新增 session/fallback_store.go，包装 RedisStore + MemoryStore 双后端 - 后台 goroutine 周期性 ping Redis，自动降级/恢复 - Store 接口新增 Metrics() 方法，MemoryStore/RedisStore/FallbackStore 均已实现 - deps_core.go 改造为始终创建 MemoryStore 兜底，Redis 启用时包装为 FallbackStore - Manager 新增 GetStoreMetrics() 暴露存储状态
141 lines
3.5 KiB
Go
141 lines
3.5 KiB
Go
package session
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// FallbackStore 会话存储降级保护层
|
||
// Redis 正常时委托给 RedisStore，异常时自动降级到 MemoryStore
|
||
// 后台健康检测 goroutine 周期性检测 Redis 状态并自动恢复
|
||
type FallbackStore struct {
|
||
redis *RedisStore
|
||
memory Store
|
||
client *redis.Client
|
||
|
||
healthy atomic.Bool // Redis 当前是否健康
|
||
degraded atomic.Bool // 是否处于降级模式（曾因异常而切换到 Memory）
|
||
}
|
||
|
||
// NewFallbackStore 创建降级保护存储
|
||
// checkInterval: 健康检测间隔，建议与 session cleanup_interval 一致
|
||
func NewFallbackStore(redis *RedisStore, memory Store, checkInterval time.Duration) *FallbackStore {
|
||
fs := &FallbackStore{
|
||
redis: redis,
|
||
memory: memory,
|
||
client: redis.client,
|
||
}
|
||
fs.healthy.Store(true)
|
||
|
||
// 后台健康检测 goroutine
|
||
go fs.healthLoop(checkInterval)
|
||
|
||
log.Println("[FallbackStore] 已启动 Redis 健康检测")
|
||
return fs
|
||
}
|
||
|
||
// healthLoop 定期 ping Redis，自动升降级
|
||
func (fs *FallbackStore) healthLoop(interval time.Duration) {
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
|
||
for range ticker.C {
|
||
prev := fs.healthy.Load()
|
||
ok := fs.ping()
|
||
fs.healthy.Store(ok)
|
||
|
||
if ok && !prev {
|
||
// Redis 恢复
|
||
fs.degraded.Store(false)
|
||
log.Println("[FallbackStore] Redis 已恢复，切回 Redis 存储")
|
||
} else if !ok && prev {
|
||
// Redis 掉线，标记降级
|
||
fs.degraded.Store(true)
|
||
log.Println("[FallbackStore] Redis 不可用，已降级为内存存储")
|
||
}
|
||
}
|
||
}
|
||
|
||
// ping 检查 Redis 连接状态（带超时）
|
||
func (fs *FallbackStore) ping() bool {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
defer cancel()
|
||
return fs.client.Ping(ctx).Err() == nil
|
||
}
|
||
|
||
// current 返回当前应使用的 Store：Redis 健康用 Redis，否则用 Memory
|
||
func (fs *FallbackStore) current() Store {
|
||
if fs.healthy.Load() {
|
||
return fs.redis
|
||
}
|
||
return fs.memory
|
||
}
|
||
|
||
// ---- Store 接口实现（委托给 current()）----
|
||
|
||
func (fs *FallbackStore) Get(id string) (*Session, error) {
|
||
return fs.current().Get(id)
|
||
}
|
||
|
||
func (fs *FallbackStore) Set(s *Session) error {
|
||
return fs.current().Set(s)
|
||
}
|
||
|
||
func (fs *FallbackStore) Delete(id string) error {
|
||
return fs.current().Delete(id)
|
||
}
|
||
|
||
func (fs *FallbackStore) DeleteByUID(uid uint) error {
|
||
return fs.current().DeleteByUID(uid)
|
||
}
|
||
|
||
func (fs *FallbackStore) DeleteByUIDExclude(uid uint, excludeSID string) error {
|
||
return fs.current().DeleteByUIDExclude(uid, excludeSID)
|
||
}
|
||
|
||
func (fs *FallbackStore) ListByUID(uid uint) ([]*Session, error) {
|
||
return fs.current().ListByUID(uid)
|
||
}
|
||
|
||
func (fs *FallbackStore) Cleanup() int {
|
||
return fs.current().Cleanup()
|
||
}
|
||
|
||
func (fs *FallbackStore) StartCleanup(interval time.Duration) {
|
||
// MemoryStore 已在外部启动 Cleanup，此处无需重复
|
||
}
|
||
|
||
// ---- Metrics（供管理面板使用）----
|
||
|
||
// Metrics 返回存储状态信息
|
||
func (fs *FallbackStore) Metrics() StoreMetrics {
|
||
healthy := fs.healthy.Load()
|
||
isDegraded := fs.degraded.Load()
|
||
|
||
m := StoreMetrics{
|
||
Type: "redis",
|
||
Healthy: healthy,
|
||
}
|
||
|
||
if !healthy {
|
||
m.Type = "memory"
|
||
m.Degraded = true
|
||
} else if isDegraded {
|
||
m.Degraded = true
|
||
m.DegradedNote = "Redis 已恢复，但存在历史降级记录（重启后清除）"
|
||
}
|
||
|
||
// 活跃会话数：Redis 模式不统计（代价高），Memory 模式可从 memory store 获取
|
||
if !healthy {
|
||
if memMetrics := fs.memory.Metrics(); memMetrics.Type == "memory" {
|
||
m.ActiveCount = memMetrics.ActiveCount
|
||
}
|
||
}
|
||
|
||
return m
|
||
}
|