feat: FallbackStore 恢复时静默迁移内存会话到 Redis

- MemoryStore 新增 Snapshot() 返回活跃会话副本
- Redis 恢复时自动将降级期间的内存会话 Pipeline 写入 Redis
- 迁移成功后再切换 healthy 标记,用户无需重新登录
- 迁移失败则保持内存模式,避免切到空的 Redis 再被迫降级
This commit is contained in:
2026-05-31 11:46:50 +08:00
parent 12de8be0cf
commit 37c83d4c12
2 changed files with 75 additions and 10 deletions

View File

@ -2,6 +2,8 @@ package session
import (
"context"
"encoding/json"
"fmt"
"log"
"sync/atomic"
"time"
@ -11,19 +13,20 @@ import (
// FallbackStore 会话存储降级保护层
// Redis 正常时委托给 RedisStore异常时自动降级到 MemoryStore
// 后台健康检测 goroutine 周期性检测 Redis 状态并自动恢复
// Redis 恢复时自动将降级期间的内存会话静默迁移回 Redis用户无感知
type FallbackStore struct {
redis *RedisStore
memory Store
memory *MemoryStore
client *redis.Client
healthy atomic.Bool // Redis 当前是否健康
degraded atomic.Bool // 是否处于降级模式(曾因异常而切换到 Memory
migrating atomic.Bool // 是否正在执行迁移(避免并发迁移)
}
// NewFallbackStore 创建降级保护存储
// checkInterval: 健康检测间隔,建议与 session cleanup_interval 一致
func NewFallbackStore(redis *RedisStore, memory Store, checkInterval time.Duration) *FallbackStore {
func NewFallbackStore(redis *RedisStore, memory *MemoryStore, checkInterval time.Duration) *FallbackStore {
fs := &FallbackStore{
redis: redis,
memory: memory,
@ -34,11 +37,11 @@ func NewFallbackStore(redis *RedisStore, memory Store, checkInterval time.Durati
// 后台健康检测 goroutine
go fs.healthLoop(checkInterval)
log.Println("[FallbackStore] 已启动 Redis 健康检测")
log.Println("[FallbackStore] 已启动 Redis 健康检测(含自动迁移)")
return fs
}
// healthLoop 定期 ping Redis自动升降级
// healthLoop 定期 ping Redis自动升降级 + 静默迁移
func (fs *FallbackStore) healthLoop(interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
@ -46,20 +49,62 @@ func (fs *FallbackStore) healthLoop(interval time.Duration) {
for range ticker.C {
prev := fs.healthy.Load()
ok := fs.ping()
fs.healthy.Store(ok)
if ok && !prev {
// Redis 恢复
// Redis 恢复:先迁移内存会话,再切回 Redis
fs.migrateSessions()
fs.healthy.Store(true)
fs.degraded.Store(false)
log.Println("[FallbackStore] Redis 已恢复,切回 Redis 存储")
} else if !ok && prev {
// Redis 掉线,标记降级
fs.healthy.Store(false)
fs.degraded.Store(true)
log.Println("[FallbackStore] Redis 不可用,已降级为内存存储")
} else {
fs.healthy.Store(ok)
}
}
}
// 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
}
log.Printf("[FallbackStore] Redis 已恢复,静默迁移 %d 条会话", len(sessions))
}
// ping 检查 Redis 连接状态(带超时)
func (fs *FallbackStore) ping() bool {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)

View File

@ -211,3 +211,23 @@ func (ms *MemoryStore) Metrics() StoreMetrics {
ActiveCount: count,
}
}
// Snapshot 返回所有活跃会话的副本(用于 Redis 恢复时迁移)
func (ms *MemoryStore) Snapshot() []*Session {
ms.mu.RLock()
defer ms.mu.RUnlock()
var active []*Session
for _, s := range ms.sessions {
timeout := ms.idleTimeout
if s.RememberMe {
timeout = ms.rememberTimeout
}
if !s.IsExpired(timeout) {
// 返回副本,避免外部修改内部 map
copyS := *s
active = append(active, &copyS)
}
}
return active
}