Victor_Jay
a9b6dddf70
- migrateSessions() 写入 Redis 后删除 MemoryStore 中对应数据 - 避免残留数据等到 TTL 自然过期才清理（最长 30 天）
274 lines
7.3 KiB
Go
274 lines
7.3 KiB
Go
package session
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// FallbackStore 会话存储降级保护层
|
||
// Redis 正常时委托给 RedisStore，异常时自动降级到 MemoryStore
|
||
// Redis 恢复时自动将降级期间的内存会话静默迁移回 Redis，用户无感知
|
||
type FallbackStore struct {
|
||
redis *RedisStore
|
||
memory *MemoryStore
|
||
client *redis.Client
|
||
|
||
healthy atomic.Bool // Redis 当前是否健康
|
||
degraded atomic.Bool // 是否处于降级模式（曾因异常而切换到 Memory）
|
||
migrating atomic.Bool // 是否正在执行迁移（避免并发迁移）
|
||
recovering atomic.Bool // 是否正在执行快速恢复轮询
|
||
}
|
||
|
||
// 健康检查间隔：正常模式下定期检测 Redis 连通性
|
||
const healthCheckInterval = 30 * time.Second
|
||
|
||
// 快速恢复间隔：降级模式下高频轮询，快速切回 Redis
|
||
const recoveryInterval = 5 * time.Second
|
||
|
||
// NewFallbackStore 创建降级保护存储
|
||
func NewFallbackStore(redis *RedisStore, memory *MemoryStore) *FallbackStore {
|
||
fs := &FallbackStore{
|
||
redis: redis,
|
||
memory: memory,
|
||
client: redis.client,
|
||
}
|
||
fs.healthy.Store(true)
|
||
|
||
// 后台健康检测 + 快速恢复（独立 goroutine 互不阻塞）
|
||
go fs.healthLoop()
|
||
go fs.recoveryLoop()
|
||
|
||
log.Println("[FallbackStore] 已启动 Redis 健康检测（健康检测 30s，降级恢复 5s，含自动迁移）")
|
||
return fs
|
||
}
|
||
|
||
// healthLoop 定期 ping Redis（30s 间隔），自动升降级 + 静默迁移
|
||
func (fs *FallbackStore) healthLoop() {
|
||
ticker := time.NewTicker(healthCheckInterval)
|
||
defer ticker.Stop()
|
||
|
||
for range ticker.C {
|
||
prev := fs.healthy.Load()
|
||
ok := fs.ping()
|
||
|
||
if ok && !prev {
|
||
// Redis 恢复：先切到 Redis（新会话立即走 Redis），再迁移内存存量
|
||
fs.healthy.Store(true)
|
||
fs.migrateSessions()
|
||
fs.degraded.Store(false)
|
||
} else if !ok && prev {
|
||
// Redis 掉线，标记降级（快速恢复由 recoveryLoop 负责）
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
log.Println("[FallbackStore] Redis 不可用，已降级为内存存储")
|
||
} else {
|
||
fs.healthy.Store(ok)
|
||
}
|
||
}
|
||
}
|
||
|
||
// recoveryLoop 启动后持续监听退化状态，一旦降级立即启动快速恢复轮询（5s 间隔）
|
||
// 退出条件：Redis 恢复成功或 connectOnce 竞争到恢复权限后执行迁移
|
||
func (fs *FallbackStore) recoveryLoop() {
|
||
ticker := time.NewTicker(recoveryInterval)
|
||
defer ticker.Stop()
|
||
|
||
for range ticker.C {
|
||
if fs.healthy.Load() {
|
||
continue // 正常模式，无需恢复
|
||
}
|
||
// 降级模式：尝试恢复
|
||
if !fs.recovering.CompareAndSwap(false, true) {
|
||
continue // 已有恢复在进行中
|
||
}
|
||
if !fs.ping() {
|
||
fs.recovering.Store(false)
|
||
continue // 仍未恢复
|
||
}
|
||
|
||
// Redis 已恢复
|
||
fs.healthy.Store(true)
|
||
fs.migrateSessions()
|
||
fs.degraded.Store(false)
|
||
fs.recovering.Store(false)
|
||
}
|
||
}
|
||
|
||
// migrateSessions 将降级期间内存中的活跃会话同步到 Redis
|
||
func (fs *FallbackStore) migrateSessions() {
|
||
if !fs.migrating.CompareAndSwap(false, true) {
|
||
return // 已有迁移在进行中
|
||
}
|
||
defer fs.migrating.Store(false)
|
||
|
||
sessions := fs.memory.Snapshot()
|
||
if len(sessions) == 0 {
|
||
log.Println("[FallbackStore] 无需迁移（降级期间无活跃会话）")
|
||
return
|
||
}
|
||
|
||
ctx := context.Background()
|
||
pipe := fs.client.Pipeline()
|
||
|
||
for _, s := range sessions {
|
||
data, err := json.Marshal(s)
|
||
if err != nil {
|
||
log.Printf("[FallbackStore] 迁移序列化失败 sid=%s err=%v", s.ID[:16]+"...", err)
|
||
continue
|
||
}
|
||
ttl := fs.redis.idleTimeout
|
||
if s.RememberMe {
|
||
ttl = fs.redis.rememberTimeout
|
||
}
|
||
pipe.Set(ctx, keyPrefix+s.ID, data, ttl)
|
||
pipe.SAdd(ctx, userKeyPrefix+fmt.Sprint(s.UserID), s.ID)
|
||
}
|
||
|
||
_, err := pipe.Exec(ctx)
|
||
if err != nil {
|
||
log.Printf("[FallbackStore] 迁移到 Redis 失败: %v，保持内存模式", err)
|
||
return
|
||
}
|
||
|
||
// 迁移成功后清理内存中的残留数据
|
||
for _, s := range sessions {
|
||
_ = fs.memory.Delete(s.ID)
|
||
}
|
||
|
||
log.Printf("[FallbackStore] Redis 已恢复，静默迁移 %d 条会话（内存已清理）", len(sessions))
|
||
}
|
||
|
||
// 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 接口实现（Redis 操作失败时立即降级到内存）----
|
||
// 不依赖周期性健康检查，操作级故障即时响应
|
||
|
||
func (fs *FallbackStore) Get(id string) (*Session, error) {
|
||
if fs.healthy.Load() {
|
||
s, err := fs.redis.Get(id)
|
||
if err == nil {
|
||
return s, nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis Get 失败，降级到内存: %v", err)
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.Get(id)
|
||
}
|
||
|
||
func (fs *FallbackStore) Set(s *Session) error {
|
||
if fs.healthy.Load() {
|
||
if err := fs.redis.Set(s); err == nil {
|
||
return nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis Set 失败，降级到内存")
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.Set(s)
|
||
}
|
||
|
||
func (fs *FallbackStore) Delete(id string) error {
|
||
if fs.healthy.Load() {
|
||
if err := fs.redis.Delete(id); err == nil {
|
||
return nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis Delete 失败，降级到内存")
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.Delete(id)
|
||
}
|
||
|
||
func (fs *FallbackStore) DeleteByUID(uid uint) error {
|
||
if fs.healthy.Load() {
|
||
if err := fs.redis.DeleteByUID(uid); err == nil {
|
||
return nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis DeleteByUID 失败，降级到内存")
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.DeleteByUID(uid)
|
||
}
|
||
|
||
func (fs *FallbackStore) DeleteByUIDExclude(uid uint, excludeSID string) error {
|
||
if fs.healthy.Load() {
|
||
if err := fs.redis.DeleteByUIDExclude(uid, excludeSID); err == nil {
|
||
return nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis DeleteByUIDExclude 失败，降级到内存")
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.DeleteByUIDExclude(uid, excludeSID)
|
||
}
|
||
|
||
func (fs *FallbackStore) ListByUID(uid uint) ([]*Session, error) {
|
||
if fs.healthy.Load() {
|
||
sessions, err := fs.redis.ListByUID(uid)
|
||
if err == nil {
|
||
return sessions, nil
|
||
}
|
||
log.Printf("[FallbackStore] Redis ListByUID 失败，降级到内存: %v", err)
|
||
fs.healthy.Store(false)
|
||
fs.degraded.Store(true)
|
||
}
|
||
return fs.memory.ListByUID(uid)
|
||
}
|
||
|
||
func (fs *FallbackStore) Cleanup() int {
|
||
if fs.healthy.Load() {
|
||
return fs.redis.Cleanup()
|
||
}
|
||
return fs.memory.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,
|
||
ActiveCount: -1, // Redis 模式不统计活跃会话数（代价高），模板 ge 0 判断自动隐藏
|
||
}
|
||
|
||
if !healthy {
|
||
m.Type = "memory"
|
||
m.Degraded = true
|
||
m.ActiveCount = fs.memory.Metrics().ActiveCount
|
||
} else if isDegraded {
|
||
m.Degraded = true
|
||
m.DegradedNote = "Redis 已恢复，但存在历史降级记录（重启后清除）"
|
||
}
|
||
|
||
return m
|
||
}
|