team/tyapi-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.1

Browse Source
This commit is contained in:
liangzai
2025-07-28 01:46:39 +08:00
parent b03129667a
commit 357639462a
219 changed files with 21634 additions and 8138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
internal/shared/cache/cache_config_manager.go vendored Normal file
View File

@@ -0,0 +1,194 @@
package cache
import (
"sync"
)
// CacheConfigManager 缓存配置管理器
// 提供全局缓存配置管理和表级别的缓存决策
type CacheConfigManager struct {
config CacheConfig
mutex sync.RWMutex
}
// GlobalCacheConfigManager 全局缓存配置管理器实例
var GlobalCacheConfigManager *CacheConfigManager
// InitCacheConfigManager 初始化全局缓存配置管理器
func InitCacheConfigManager(config CacheConfig) {
GlobalCacheConfigManager = &CacheConfigManager{
config: config,
}
}
// GetCacheConfig 获取当前缓存配置
func (m *CacheConfigManager) GetCacheConfig() CacheConfig {
m.mutex.RLock()
defer m.mutex.RUnlock()
return m.config
}
// UpdateCacheConfig 更新缓存配置
func (m *CacheConfigManager) UpdateCacheConfig(config CacheConfig) {
m.mutex.Lock()
defer m.mutex.Unlock()
m.config = config
}
// IsTableCacheEnabled 检查表是否启用缓存
func (m *CacheConfigManager) IsTableCacheEnabled(tableName string) bool {
m.mutex.RLock()
defer m.mutex.RUnlock()
// 检查表是否在禁用列表中
for _, disabledTable := range m.config.DisabledTables {
if disabledTable == tableName {
return false
}
}
// 如果配置了启用列表，只对启用列表中的表启用缓存
if len(m.config.EnabledTables) > 0 {
for _, enabledTable := range m.config.EnabledTables {
if enabledTable == tableName {
return true
}
}
return false
}
// 如果没有配置启用列表，默认对所有表启用缓存（除了禁用列表中的表）
return true
}
// IsTableCacheDisabled 检查表是否禁用缓存
func (m *CacheConfigManager) IsTableCacheDisabled(tableName string) bool {
return !m.IsTableCacheEnabled(tableName)
}
// GetEnabledTables 获取启用缓存的表列表
func (m *CacheConfigManager) GetEnabledTables() []string {
m.mutex.RLock()
defer m.mutex.RUnlock()
return m.config.EnabledTables
}
// GetDisabledTables 获取禁用缓存的表列表
func (m *CacheConfigManager) GetDisabledTables() []string {
m.mutex.RLock()
defer m.mutex.RUnlock()
return m.config.DisabledTables
}
// AddEnabledTable 添加启用缓存的表
func (m *CacheConfigManager) AddEnabledTable(tableName string) {
m.mutex.Lock()
defer m.mutex.Unlock()
// 检查是否已存在
for _, table := range m.config.EnabledTables {
if table == tableName {
return
}
}
m.config.EnabledTables = append(m.config.EnabledTables, tableName)
}
// AddDisabledTable 添加禁用缓存的表
func (m *CacheConfigManager) AddDisabledTable(tableName string) {
m.mutex.Lock()
defer m.mutex.Unlock()
// 检查是否已存在
for _, table := range m.config.DisabledTables {
if table == tableName {
return
}
}
m.config.DisabledTables = append(m.config.DisabledTables, tableName)
}
// RemoveEnabledTable 移除启用缓存的表
func (m *CacheConfigManager) RemoveEnabledTable(tableName string) {
m.mutex.Lock()
defer m.mutex.Unlock()
var newEnabledTables []string
for _, table := range m.config.EnabledTables {
if table != tableName {
newEnabledTables = append(newEnabledTables, table)
}
}
m.config.EnabledTables = newEnabledTables
}
// RemoveDisabledTable 移除禁用缓存的表
func (m *CacheConfigManager) RemoveDisabledTable(tableName string) {
m.mutex.Lock()
defer m.mutex.Unlock()
var newDisabledTables []string
for _, table := range m.config.DisabledTables {
if table != tableName {
newDisabledTables = append(newDisabledTables, table)
}
}
m.config.DisabledTables = newDisabledTables
}
// GetTableCacheStatus 获取表的缓存状态信息
func (m *CacheConfigManager) GetTableCacheStatus(tableName string) map[string]interface{} {
m.mutex.RLock()
defer m.mutex.RUnlock()
status := map[string]interface{}{
"table_name": tableName,
"enabled": m.IsTableCacheEnabled(tableName),
"disabled": m.IsTableCacheDisabled(tableName),
}
// 检查是否在启用列表中
for _, table := range m.config.EnabledTables {
if table == tableName {
status["in_enabled_list"] = true
break
}
}
// 检查是否在禁用列表中
for _, table := range m.config.DisabledTables {
if table == tableName {
status["in_disabled_list"] = true
break
}
}
return status
}
// GetAllTableStatus 获取所有表的缓存状态
func (m *CacheConfigManager) GetAllTableStatus() map[string]interface{} {
m.mutex.RLock()
defer m.mutex.RUnlock()
result := map[string]interface{}{
"enabled_tables": m.config.EnabledTables,
"disabled_tables": m.config.DisabledTables,
"config": map[string]interface{}{
"default_ttl": m.config.DefaultTTL,
"table_prefix": m.config.TablePrefix,
"max_cache_size": m.config.MaxCacheSize,
"cache_complex_sql": m.config.CacheComplexSQL,
"enable_stats": m.config.EnableStats,
"enable_warmup": m.config.EnableWarmup,
"penetration_guard": m.config.PenetrationGuard,
"bloom_filter": m.config.BloomFilter,
"auto_invalidate": m.config.AutoInvalidate,
"invalidate_delay": m.config.InvalidateDelay,
},
}
return result
}

240
internal/shared/cache/gorm_cache_plugin.go vendored
View File

@@ -8,6 +8,7 @@ import (
"fmt"
"reflect"
"strings"
"sync"
"time"
"go.uber.org/zap"
@@ -21,6 +22,10 @@ type GormCachePlugin struct {
cache interfaces.CacheService
logger *zap.Logger
config CacheConfig
// 缓存失效去重机制
invalidationQueue map[string]*time.Timer
queueMutex sync.RWMutex
}
// CacheConfig 缓存配置
@@ -58,7 +63,8 @@ func DefaultCacheConfig() CacheConfig {
PenetrationGuard: true,
BloomFilter: false,
AutoInvalidate: true,
InvalidateDelay: 100 * time.Millisecond,
// 增加延迟失效时间，减少频繁的缓存失效操作
InvalidateDelay: 500 * time.Millisecond,
}
}
@@ -73,6 +79,7 @@ func NewGormCachePlugin(cache interfaces.CacheService, logger *zap.Logger, confi
cache: cache,
logger: logger,
config: cfg,
invalidationQueue: make(map[string]*time.Timer),
}
}
@@ -87,12 +94,30 @@ func (p *GormCachePlugin) Initialize(db *gorm.DB) error {
zap.Duration("default_ttl", p.config.DefaultTTL),
zap.Bool("auto_invalidate", p.config.AutoInvalidate),
zap.Bool("penetration_guard", p.config.PenetrationGuard),
zap.Duration("invalidate_delay", p.config.InvalidateDelay),
)
// 注册回调函数
return p.registerCallbacks(db)
}
// Shutdown 关闭插件，清理资源
func (p *GormCachePlugin) Shutdown() {
p.queueMutex.Lock()
defer p.queueMutex.Unlock()
// 停止所有定时器
for table, timer := range p.invalidationQueue {
timer.Stop()
p.logger.Debug("停止缓存失效定时器", zap.String("table", table))
}
// 清空队列
p.invalidationQueue = make(map[string]*time.Timer)
p.logger.Info("GORM缓存插件已关闭")
}
// registerCallbacks 注册GORM回调
func (p *GormCachePlugin) registerCallbacks(db *gorm.DB) error {
// Query回调 - 查询时检查缓存
@@ -117,6 +142,7 @@ func (p *GormCachePlugin) registerCallbacks(db *gorm.DB) error {
func (p *GormCachePlugin) beforeQuery(db *gorm.DB) {
// 检查是否启用缓存
if !p.shouldCache(db) {
p.logger.Debug("跳过缓存", zap.String("table", db.Statement.Table))
return
}
@@ -147,7 +173,15 @@ func (p *GormCachePlugin) beforeQuery(db *gorm.DB) {
p.updateStats("hit", db.Statement.Table)
}
return
} else {
p.logger.Warn("缓存数据恢复失败，将执行数据库查询",
zap.String("cache_key", cacheKey),
zap.Error(err))
}
} else {
p.logger.Debug("缓存未命中",
zap.String("cache_key", cacheKey),
zap.Error(err))
}
// 缓存未命中，设置标记
@@ -195,7 +229,10 @@ func (p *GormCachePlugin) afterCreate(db *gorm.DB) {
return
}
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
// 只对启用缓存的表执行失效操作
if p.shouldInvalidateTable(db.Statement.Table) {
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
}
}
// afterUpdate 更新后回调
@@ -204,7 +241,10 @@ func (p *GormCachePlugin) afterUpdate(db *gorm.DB) {
return
}
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
// 只对启用缓存的表执行失效操作
if p.shouldInvalidateTable(db.Statement.Table) {
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
}
}
// afterDelete 删除后回调
@@ -213,26 +253,64 @@ func (p *GormCachePlugin) afterDelete(db *gorm.DB) {
return
}
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
// 只对启用缓存的表执行失效操作
if p.shouldInvalidateTable(db.Statement.Table) {
p.invalidateTableCache(db.Statement.Context, db.Statement.Table)
}
}
// ================ 缓存管理方法 ================
// shouldInvalidateTable 判断是否应该对表执行缓存失效操作
func (p *GormCachePlugin) shouldInvalidateTable(table string) bool {
// 使用全局缓存配置管理器进行智能决策
if GlobalCacheConfigManager != nil {
return GlobalCacheConfigManager.IsTableCacheEnabled(table)
}
// 如果全局管理器未初始化，使用本地配置
// 检查表是否在禁用列表中
for _, disabledTable := range p.config.DisabledTables {
if disabledTable == table {
return false
}
}
// 如果配置了启用列表，只对启用列表中的表执行失效操作
if len(p.config.EnabledTables) > 0 {
for _, enabledTable := range p.config.EnabledTables {
if enabledTable == table {
return true
}
}
return false
}
// 如果没有配置启用列表，默认对所有表执行失效操作（除了禁用列表中的表）
return true
}
// shouldCache 判断是否应该缓存
func (p *GormCachePlugin) shouldCache(db *gorm.DB) bool {
// 检查是否明确禁用缓存
// 检查是否手动禁用缓存
if value, ok := db.Statement.Get("cache:disabled"); ok && value.(bool) {
return false
}
// 检查是否明确启用缓存
// 检查是否手动启用缓存
if value, ok := db.Statement.Get("cache:enabled"); ok && value.(bool) {
return true
}
// 使用全局缓存配置管理器进行智能决策
if GlobalCacheConfigManager != nil {
return GlobalCacheConfigManager.IsTableCacheEnabled(db.Statement.Table)
}
// 如果全局管理器未初始化，使用本地配置
// 检查表是否在禁用列表中
for _, table := range p.config.DisabledTables {
if table == db.Statement.Table {
for _, disabledTable := range p.config.DisabledTables {
if disabledTable == db.Statement.Table {
return false
}
}
@@ -247,11 +325,7 @@ func (p *GormCachePlugin) shouldCache(db *gorm.DB) bool {
return false
}
// 检查是否为复杂查询
if !p.config.CacheComplexSQL && p.isComplexQuery(db) {
return false
}
// 如果没有配置启用列表，默认对所有表启用缓存（除了禁用列表中的表）
return true
}
@@ -292,10 +366,27 @@ func (p *GormCachePlugin) generateCacheKey(db *gorm.DB) string {
// hashSQL 对SQL语句和参数进行hash
func (p *GormCachePlugin) hashSQL(sql string, vars []interface{}) string {
// 将SQL和参数组合
// 修复：改进SQL hash生成，确保唯一性
combined := sql
for _, v := range vars {
combined += fmt.Sprintf(":%v", v)
// 使用更精确的格式化，避免类型信息丢失
switch val := v.(type) {
case string:
combined += ":" + val
case int, int32, int64:
combined += fmt.Sprintf(":%d", val)
case float32, float64:
combined += fmt.Sprintf(":%f", val)
case bool:
combined += fmt.Sprintf(":%t", val)
default:
// 对于复杂类型，使用JSON序列化
if jsonData, err := json.Marshal(v); err == nil {
combined += ":" + string(jsonData)
} else {
combined += fmt.Sprintf(":%v", v)
}
}
}
// 计算MD5 hash
@@ -328,9 +419,22 @@ func (p *GormCachePlugin) saveToCache(ctx context.Context, cacheKey string, db *
return fmt.Errorf("查询结果为空")
}
// 修复：改进缓存数据保存逻辑
var dataToCache interface{}
// 如果dest是切片，需要特殊处理
destValue := reflect.ValueOf(dest)
if destValue.Kind() == reflect.Ptr && destValue.Elem().Kind() == reflect.Slice {
// 对于切片，直接使用原始数据
dataToCache = destValue.Elem().Interface()
} else {
// 对于单个对象，也直接使用原始数据
dataToCache = dest
}
// 构建缓存结果
result := CachedResult{
Data: dest,
Data: dataToCache,
RowCount: db.Statement.RowsAffected,
Timestamp: time.Now(),
}
@@ -340,6 +444,10 @@ func (p *GormCachePlugin) saveToCache(ctx context.Context, cacheKey string, db *
// 保存到缓存
if err := p.cache.Set(ctx, cacheKey, result, ttl); err != nil {
p.logger.Error("保存查询结果到缓存失败",
zap.String("cache_key", cacheKey),
zap.Error(err),
)
return fmt.Errorf("保存到缓存失败: %w", err)
}
@@ -364,25 +472,35 @@ func (p *GormCachePlugin) restoreFromCache(db *gorm.DB, cachedResult *CachedResu
return fmt.Errorf("目标对象必须是指针")
}
// 将缓存数据复制到目标
// 修复：改进缓存数据恢复逻辑
cachedValue := reflect.ValueOf(cachedResult.Data)
if !cachedValue.Type().AssignableTo(destValue.Elem().Type()) {
// 如果类型完全匹配，直接赋值
if cachedValue.Type().AssignableTo(destValue.Elem().Type()) {
destValue.Elem().Set(cachedValue)
} else {
// 尝试JSON转换
jsonData, err := json.Marshal(cachedResult.Data)
if err != nil {
return fmt.Errorf("缓存数据类型不匹配")
p.logger.Error("序列化缓存数据失败", zap.Error(err))
return fmt.Errorf("缓存数据类型转换失败: %w", err)
}
if err := json.Unmarshal(jsonData, db.Statement.Dest); err != nil {
p.logger.Error("反序列化缓存数据失败",
zap.String("json_data", string(jsonData)),
zap.Error(err))
return fmt.Errorf("JSON反序列化失败: %w", err)
}
} else {
destValue.Elem().Set(cachedValue)
}
// 设置影响行数
db.Statement.RowsAffected = cachedResult.RowCount
p.logger.Debug("从缓存恢复数据成功",
zap.Int64("rows", cachedResult.RowCount),
zap.Time("timestamp", cachedResult.Timestamp))
return nil
}
@@ -400,37 +518,79 @@ func (p *GormCachePlugin) getTTL(db *gorm.DB) time.Duration {
// invalidateTableCache 失效表相关缓存
func (p *GormCachePlugin) invalidateTableCache(ctx context.Context, table string) {
if ctx == nil {
ctx = context.Background()
}
// 延迟失效（避免并发问题）
if p.config.InvalidateDelay > 0 {
time.AfterFunc(p.config.InvalidateDelay, func() {
p.doInvalidateTableCache(ctx, table)
})
} else {
p.doInvalidateTableCache(ctx, table)
// 使用去重机制，避免重复的缓存失效操作
p.queueMutex.Lock()
defer p.queueMutex.Unlock()
// 如果已经有相同的失效操作在队列中，取消之前的定时器
if timer, exists := p.invalidationQueue[table]; exists {
timer.Stop()
delete(p.invalidationQueue, table)
}
// 创建独立的上下文，避免受到原始请求上下文的影响
// 设置合理的超时时间，避免缓存失效操作阻塞
cacheCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
// 创建新的定时器
timer := time.AfterFunc(p.config.InvalidateDelay, func() {
// 执行缓存失效
p.doInvalidateTableCache(cacheCtx, table)
// 清理定时器引用
p.queueMutex.Lock()
delete(p.invalidationQueue, table)
p.queueMutex.Unlock()
// 取消上下文
cancel()
})
// 将定时器加入队列
p.invalidationQueue[table] = timer
p.logger.Debug("缓存失效操作已加入队列",
zap.String("table", table),
zap.Duration("delay", p.config.InvalidateDelay),
)
}
// doInvalidateTableCache 执行缓存失效
func (p *GormCachePlugin) doInvalidateTableCache(ctx context.Context, table string) {
pattern := fmt.Sprintf("%s:%s:*", p.config.TablePrefix, table)
if err := p.cache.DeletePattern(ctx, pattern); err != nil {
p.logger.Warn("失效表缓存失败",
// 添加重试机制，提高缓存失效的可靠性
maxRetries := 3
for attempt := 1; attempt <= maxRetries; attempt++ {
if err := p.cache.DeletePattern(ctx, pattern); err != nil {
if attempt < maxRetries {
p.logger.Warn("缓存失效失败，准备重试",
zap.String("table", table),
zap.String("pattern", pattern),
zap.Int("attempt", attempt),
zap.Error(err),
)
// 短暂延迟后重试
time.Sleep(time.Duration(attempt) * 100 * time.Millisecond)
continue
}
p.logger.Warn("失效表缓存失败",
zap.String("table", table),
zap.String("pattern", pattern),
zap.Int("attempts", maxRetries),
zap.Error(err),
)
return
}
// 成功删除，记录日志并退出
p.logger.Debug("表缓存已失效",
zap.String("table", table),
zap.String("pattern", pattern),
zap.Error(err),
)
return
}
p.logger.Debug("表缓存已失效",
zap.String("table", table),
zap.String("pattern", pattern),
)
}
// updateStats 更新统计信息