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

fix

Browse Source
This commit is contained in:
liangzai
2025-07-31 15:41:00 +08:00
parent f3a3bc84c7
commit 934dce2776
36 changed files with 1614 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,7 +22,7 @@ type GormCachePlugin struct {
cache interfaces.CacheService
logger *zap.Logger
config CacheConfig
// 缓存失效去重机制
invalidationQueue map[string]*time.Timer
queueMutex sync.RWMutex
@@ -64,7 +64,7 @@ func DefaultCacheConfig() CacheConfig {
BloomFilter: false,
AutoInvalidate: true,
// 增加延迟失效时间，减少频繁的缓存失效操作
InvalidateDelay: 500 * time.Millisecond,
InvalidateDelay: 500 * time.Millisecond,
}
}
@@ -76,9 +76,9 @@ func NewGormCachePlugin(cache interfaces.CacheService, logger *zap.Logger, confi
}
return &GormCachePlugin{
cache: cache,
logger: logger,
config: cfg,
cache: cache,
logger: logger,
config: cfg,
invalidationQueue: make(map[string]*time.Timer),
}
}
@@ -105,16 +105,16 @@ func (p *GormCachePlugin) Initialize(db *gorm.DB) error {
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缓存插件已关闭")
}
@@ -174,13 +174,13 @@ func (p *GormCachePlugin) beforeQuery(db *gorm.DB) {
}
return
} else {
p.logger.Warn("缓存数据恢复失败，将执行数据库查询",
zap.String("cache_key", cacheKey),
p.logger.Warn("缓存数据恢复失败，将执行数据库查询",
zap.String("cache_key", cacheKey),
zap.Error(err))
}
} else {
p.logger.Debug("缓存未命中",
zap.String("cache_key", cacheKey),
p.logger.Debug("缓存未命中",
zap.String("cache_key", cacheKey),
zap.Error(err))
}
@@ -421,7 +421,7 @@ func (p *GormCachePlugin) saveToCache(ctx context.Context, cacheKey string, db *
// 修复：改进缓存数据保存逻辑
var dataToCache interface{}
// 如果dest是切片，需要特殊处理
destValue := reflect.ValueOf(dest)
if destValue.Kind() == reflect.Ptr && destValue.Elem().Kind() == reflect.Slice {
@@ -474,7 +474,7 @@ func (p *GormCachePlugin) restoreFromCache(db *gorm.DB, cachedResult *CachedResu
// 修复：改进缓存数据恢复逻辑
cachedValue := reflect.ValueOf(cachedResult.Data)
// 如果类型完全匹配，直接赋值
if cachedValue.Type().AssignableTo(destValue.Elem().Type()) {
destValue.Elem().Set(cachedValue)
@@ -487,8 +487,8 @@ func (p *GormCachePlugin) restoreFromCache(db *gorm.DB, cachedResult *CachedResu
}
if err := json.Unmarshal(jsonData, db.Statement.Dest); err != nil {
p.logger.Error("反序列化缓存数据失败",
zap.String("json_data", string(jsonData)),
p.logger.Error("反序列化缓存数据失败",
zap.String("json_data", string(jsonData)),
zap.Error(err))
return fmt.Errorf("JSON反序列化失败: %w", err)
}
@@ -497,7 +497,7 @@ func (p *GormCachePlugin) restoreFromCache(db *gorm.DB, cachedResult *CachedResu
// 设置影响行数
db.Statement.RowsAffected = cachedResult.RowCount
p.logger.Debug("从缓存恢复数据成功",
p.logger.Debug("从缓存恢复数据成功",
zap.Int64("rows", cachedResult.RowCount),
zap.Time("timestamp", cachedResult.Timestamp))
@@ -521,34 +521,34 @@ func (p *GormCachePlugin) invalidateTableCache(ctx context.Context, table string
// 使用去重机制，避免重复的缓存失效操作
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),
@@ -574,7 +574,7 @@ func (p *GormCachePlugin) doInvalidateTableCache(ctx context.Context, table stri
time.Sleep(time.Duration(attempt) * 100 * time.Millisecond)
continue
}
p.logger.Warn("失效表缓存失败",
zap.String("table", table),
zap.String("pattern", pattern),
@@ -583,7 +583,7 @@ func (p *GormCachePlugin) doInvalidateTableCache(ctx context.Context, table stri
)
return
}
// 成功删除，记录日志并退出
p.logger.Debug("表缓存已失效",
zap.String("table", table),

36
internal/shared/database/cached_base_repository.go
View File

@@ -35,9 +35,9 @@ func (r *CachedBaseRepositoryImpl) isTableCacheEnabled() bool {
if cache.GlobalCacheConfigManager != nil {
return cache.GlobalCacheConfigManager.IsTableCacheEnabled(r.tableName)
}
// 如果全局管理器未初始化，默认启用缓存
r.logger.Warn("全局缓存配置管理器未初始化，默认启用缓存",
r.logger.Warn("全局缓存配置管理器未初始化，默认启用缓存",
zap.String("table", r.tableName))
return true
}
@@ -46,7 +46,7 @@ func (r *CachedBaseRepositoryImpl) isTableCacheEnabled() bool {
func (r *CachedBaseRepositoryImpl) shouldUseCacheForTable() bool {
// 检查表是否启用缓存
if !r.isTableCacheEnabled() {
r.logger.Debug("表未启用缓存，跳过缓存操作",
r.logger.Debug("表未启用缓存，跳过缓存操作",
zap.String("table", r.tableName))
return false
}
@@ -58,17 +58,17 @@ func (r *CachedBaseRepositoryImpl) shouldUseCacheForTable() bool {
// GetWithCache 带缓存的单条查询（智能决策）
func (r *CachedBaseRepositoryImpl) GetWithCache(ctx context.Context, dest interface{}, ttl time.Duration, where string, args ...interface{}) error {
db := r.GetDB(ctx)
// 智能决策：根据表配置决定是否使用缓存
if r.shouldUseCacheForTable() {
db = db.Set("cache:enabled", true).Set("cache:ttl", ttl)
r.logger.Debug("执行带缓存查询",
r.logger.Debug("执行带缓存查询",
zap.String("table", r.tableName),
zap.Duration("ttl", ttl),
zap.String("where", where))
} else {
db = db.Set("cache:disabled", true)
r.logger.Debug("执行无缓存查询",
r.logger.Debug("执行无缓存查询",
zap.String("table", r.tableName),
zap.String("where", where))
}
@@ -79,17 +79,17 @@ func (r *CachedBaseRepositoryImpl) GetWithCache(ctx context.Context, dest interf
// FindWithCache 带缓存的多条查询（智能决策）
func (r *CachedBaseRepositoryImpl) FindWithCache(ctx context.Context, dest interface{}, ttl time.Duration, where string, args ...interface{}) error {
db := r.GetDB(ctx)
// 智能决策：根据表配置决定是否使用缓存
if r.shouldUseCacheForTable() {
db = db.Set("cache:enabled", true).Set("cache:ttl", ttl)
r.logger.Debug("执行带缓存批量查询",
r.logger.Debug("执行带缓存批量查询",
zap.String("table", r.tableName),
zap.Duration("ttl", ttl),
zap.String("where", where))
} else {
db = db.Set("cache:disabled", true)
r.logger.Debug("执行无缓存批量查询",
r.logger.Debug("执行无缓存批量查询",
zap.String("table", r.tableName),
zap.String("where", where))
}
@@ -100,17 +100,17 @@ func (r *CachedBaseRepositoryImpl) FindWithCache(ctx context.Context, dest inter
// CountWithCache 带缓存的计数查询（智能决策）
func (r *CachedBaseRepositoryImpl) CountWithCache(ctx context.Context, count *int64, ttl time.Duration, entity interface{}, where string, args ...interface{}) error {
db := r.GetDB(ctx).Model(entity)
// 智能决策：根据表配置决定是否使用缓存
if r.shouldUseCacheForTable() {
db = db.Set("cache:enabled", true).Set("cache:ttl", ttl)
r.logger.Debug("执行带缓存计数查询",
r.logger.Debug("执行带缓存计数查询",
zap.String("table", r.tableName),
zap.Duration("ttl", ttl),
zap.String("where", where))
} else {
db = db.Set("cache:disabled", true)
r.logger.Debug("执行无缓存计数查询",
r.logger.Debug("执行无缓存计数查询",
zap.String("table", r.tableName),
zap.String("where", where))
}
@@ -121,16 +121,16 @@ func (r *CachedBaseRepositoryImpl) CountWithCache(ctx context.Context, count *in
// ListWithCache 带缓存的列表查询（智能决策）
func (r *CachedBaseRepositoryImpl) ListWithCache(ctx context.Context, dest interface{}, ttl time.Duration, options CacheListOptions) error {
db := r.GetDB(ctx)
// 智能决策：根据表配置决定是否使用缓存
if r.shouldUseCacheForTable() {
db = db.Set("cache:enabled", true).Set("cache:ttl", ttl)
r.logger.Debug("执行带缓存列表查询",
r.logger.Debug("执行带缓存列表查询",
zap.String("table", r.tableName),
zap.Duration("ttl", ttl))
} else {
db = db.Set("cache:disabled", true)
r.logger.Debug("执行无缓存列表查询",
r.logger.Debug("执行无缓存列表查询",
zap.String("table", r.tableName))
}
@@ -214,10 +214,10 @@ func (r *CachedBaseRepositoryImpl) WithLongCache() *CachedBaseRepositoryImpl {
// SmartGetByID 智能ID查询（自动缓存）
func (r *CachedBaseRepositoryImpl) SmartGetByID(ctx context.Context, id string, dest interface{}) error {
r.logger.Debug("执行智能ID查询",
r.logger.Debug("执行智能ID查询",
zap.String("table", r.tableName),
zap.String("id", id))
return r.GetWithCache(ctx, dest, 30*time.Minute, "id = ?", id)
}
@@ -237,7 +237,7 @@ func (r *CachedBaseRepositoryImpl) SmartList(ctx context.Context, dest interface
cacheTTL := r.calculateCacheTTL(options)
useCache := r.shouldUseCache(options)
r.logger.Debug("执行智能列表查询",
r.logger.Debug("执行智能列表查询",
zap.String("table", r.tableName),
zap.Bool("use_cache", useCache),
zap.Duration("cache_ttl", cacheTTL))