diff --git a/internal/fqapi/audio_proxy.go b/internal/fqapi/audio_proxy.go index 73a34d9..018ef1c 100644 --- a/internal/fqapi/audio_proxy.go +++ b/internal/fqapi/audio_proxy.go @@ -35,8 +35,6 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri // 存储到数据库 if cacheDB != nil { - cacheMutex.Lock() - defer cacheMutex.Unlock() _, err := cacheDB.Exec(q(` INSERT INTO audio_cache (key, original_url, device_id, headers, created_at) @@ -71,12 +69,10 @@ func HandleAudioProxy(key string) ([]byte, string, error) { var headersJSON string var createdAt int64 - cacheMutex.Lock() err := cacheDB.QueryRow(q(` SELECT original_url, device_id, headers, created_at FROM audio_cache WHERE key = ? `), key).Scan(&originalURL, &deviceID, &headersJSON, &createdAt) - cacheMutex.Unlock() if err == sql.ErrNoRows { return nil, "", fmt.Errorf("音频信息不存在,请先调用 /audio 接口") @@ -88,9 +84,7 @@ func HandleAudioProxy(key string) ([]byte, string, error) { // 检查缓存是否过期(2小时,匹配CDN链接有效期) if time.Now().Unix()-createdAt > 7200 { // 2小时 = 7200秒 // 清理过期缓存 - cacheMutex.Lock() cacheDB.Exec(q("DELETE FROM audio_cache WHERE key = ?"), key) - cacheMutex.Unlock() return nil, "", fmt.Errorf("音频信息已过期,请重新调用 /audio 接口") } @@ -121,9 +115,6 @@ func CleanExpiredAudioCache() { return } - cacheMutex.Lock() - defer cacheMutex.Unlock() - // 删除2小时前的缓存 result, err := cacheDB.Exec(q("DELETE FROM audio_cache WHERE created_at < ?"), time.Now().Unix()-7200) if err != nil { diff --git a/internal/fqapi/cache_manager.go b/internal/fqapi/cache_manager.go index fb66ed3..98afe75 100644 --- a/internal/fqapi/cache_manager.go +++ b/internal/fqapi/cache_manager.go @@ -10,7 +10,7 @@ import ( "net/url" "strconv" "strings" - "sync" + "sync/atomic" "time" _ "github.com/jackc/pgx/v5/stdlib" @@ -22,10 +22,8 @@ var ( cacheDB *sql.DB dbDriver string // "sqlite3" | "pgx" redisCache *redis.Client - cacheMutex sync.Mutex cacheLogger = common.NewLogger("cache_manager", "INFO", "") - contentRequestCount int // content接口访问计数 - countMutex sync.Mutex + contentRequestCount atomic.Int64 // content接口访问计数 cacheCleanupTicker *time.Ticker // 缓存清理定时器 ) @@ -49,7 +47,7 @@ func q(query string) string { // InitRedis 初始化 Redis 缓存(启用后正文缓存与音频代理缓存走 Redis) func InitRedis() error { - cfg := config.GetConfig() + cfg := config.GetAppConfig() if cfg == nil || !cfg.Cache.Redis.Enabled { return nil } @@ -90,7 +88,7 @@ func InitDB() error { return nil // 已初始化 } - cfg := config.GetConfig() + cfg := config.GetAppConfig() backend := "sqlite" dsn := "./cache.db" if cfg != nil { @@ -250,7 +248,7 @@ func createTables() error { // isCacheEnabled 检查正文内容缓存是否启用 func isCacheEnabled() bool { - cfg := config.GetConfig() + cfg := config.GetAppConfig() if cfg == nil { return false } @@ -271,9 +269,6 @@ func GetCache(key string) string { return "" } - cacheMutex.Lock() - defer cacheMutex.Unlock() - var value string err := cacheDB.QueryRow(q("SELECT value FROM cache WHERE key = ?"), key).Scan(&value) if err != nil { @@ -316,9 +311,6 @@ func UpdateCache(key string, value string, expi int) error { return nil } - cacheMutex.Lock() - defer cacheMutex.Unlock() - cacheData := CacheData{ Data: value, Expi: float64(time.Now().Unix() + int64(expi)), @@ -341,7 +333,6 @@ func UpdateCache(key string, value string, expi int) error { return nil } - // CloseCache 关闭缓存数据库 func CloseCache() { if cacheDB != nil { @@ -349,18 +340,12 @@ func CloseCache() { } } - - - // CleanupAllCache 立即清理所有正文缓存 func CleanupAllCache() error { if cacheDB == nil { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - // 删除所有缓存数据 _, err := cacheDB.Exec("DELETE FROM cache") if err != nil { @@ -401,7 +386,7 @@ func CleanupAllCache() error { // CleanupExpiredCache 清理过期的正文缓存 func CleanupExpiredCache() error { - cfg := config.GetConfig() + cfg := config.GetAppConfig() if cfg == nil { return fmt.Errorf("配置未加载") } @@ -420,9 +405,6 @@ func CleanupExpiredCache() error { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - // 计算过期时间点(当前时间 - cleanup_seconds) expiredTime := time.Now().Unix() - int64(cfg.Cache.CleanupSeconds) @@ -446,7 +428,7 @@ func CleanupExpiredCache() error { // StartCacheCleanup 启动缓存自动清理任务 func StartCacheCleanup() { - cfg := config.GetConfig() + cfg := config.GetAppConfig() if cfg == nil || !cfg.Cache.CleanupEnabled || cfg.Cache.CleanupSeconds <= 0 { cacheLogger.Info("缓存自动清理未启用或配置无效,跳过启动") return @@ -488,36 +470,21 @@ func StopCacheCleanup() { } // IncrementContentRequestCount 增加content接口访问计数 -func IncrementContentRequestCount() int { - countMutex.Lock() - defer countMutex.Unlock() - contentRequestCount++ - return contentRequestCount -} +func IncrementContentRequestCount() int { return int(contentRequestCount.Add(1)) } // ResetContentRequestCount 重置访问计数 -func ResetContentRequestCount() { - countMutex.Lock() - defer countMutex.Unlock() - contentRequestCount = 0 -} +func ResetContentRequestCount() { contentRequestCount.Store(0) } // GetContentRequestCount 获取当前访问计数 -func GetContentRequestCount() int { - countMutex.Lock() - defer countMutex.Unlock() - return contentRequestCount -} +func GetContentRequestCount() int { return int(contentRequestCount.Load()) } // VacuumDatabase 压缩数据库,释放磁盘空间 +// ponytail: 删了全局串行锁,VACUUM 与写操作的并发冲突由 busy_timeout(5s) 兜底,失败仅记日志 func VacuumDatabase() error { if cacheDB == nil { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - cacheLogger.Info("开始执行数据库压缩...") if dbDriver == "pgx" { cacheLogger.Info("PostgreSQL 由 autovacuum 自动管理,无需手动压缩") @@ -539,9 +506,6 @@ func GetDatabaseStats() (map[string]interface{}, error) { return nil, fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - stats := make(map[string]interface{}) // 获取cache表记录数 @@ -596,9 +560,6 @@ func SaveStatistics(stats map[string]int64) error { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - _, err := cacheDB.Exec(q(` UPDATE statistics SET total_requests = ?, @@ -630,9 +591,6 @@ func LoadStatistics() (map[string]int64, error) { return nil, fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() - var total, search, detail, catalog, content, manga, audio, video, category int64 err := cacheDB.QueryRow(` diff --git a/internal/fqapi/cache_manager_test.go b/internal/fqapi/cache_manager_test.go new file mode 100644 index 0000000..e7846a0 --- /dev/null +++ b/internal/fqapi/cache_manager_test.go @@ -0,0 +1,57 @@ +package fqapi + +import ( + "database/sql" + "fanqie/internal/config" + "fmt" + "sync" + "testing" + + _ "github.com/mattn/go-sqlite3" +) + +// TestConcurrentCacheAccess 验证删除全局串行锁后并发读写缓存不死锁、不 panic、数据正确(配合 -race) +func TestConcurrentCacheAccess(t *testing.T) { + if err := config.LoadConfig("../../config/config.yaml"); err != nil { + t.Skipf("无法加载配置: %v", err) + } + + db, err := sql.Open("sqlite3", t.TempDir()+"/test.db?_journal_mode=WAL&_busy_timeout=5000") + if err != nil { + t.Fatal(err) + } + oldDB, oldDriver := cacheDB, dbDriver + cacheDB, dbDriver = db, "sqlite3" + defer func() { + cacheDB, dbDriver = oldDB, oldDriver + db.Close() + }() + + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS cache ( + key TEXT PRIMARY KEY, + value TEXT, + created_at INTEGER DEFAULT (strftime('%s', 'now')) + )`); err != nil { + t.Fatal(err) + } + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + for j := 0; j < 30; j++ { + key := fmt.Sprintf("k-%d-%d", i, j) + if err := UpdateCache(key, "value-"+key, 60); err != nil { + t.Errorf("UpdateCache(%s): %v", key, err) + return + } + if got := GetCache(key); got != "value-"+key { + t.Errorf("GetCache(%s) = %q, want %q", key, got, "value-"+key) + return + } + } + }(i) + } + wg.Wait() +} diff --git a/internal/fqapi/device_key_pool.go b/internal/fqapi/device_key_pool.go index a96636e..4afd5e4 100644 --- a/internal/fqapi/device_key_pool.go +++ b/internal/fqapi/device_key_pool.go @@ -399,8 +399,6 @@ func (p *DeviceKeyPool) saveDeviceToDB(item *DeviceKeyItem) error { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() // RETURNING id 同时兼容 SQLite(3.35+) 与 PostgreSQL,替代 LastInsertId() err := cacheDB.QueryRow(q(` @@ -419,8 +417,6 @@ func (p *DeviceKeyPool) updateDeviceInDB(item *DeviceKeyItem) error { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() _, err := cacheDB.Exec(q(` UPDATE device_pool @@ -438,8 +434,6 @@ func (p *DeviceKeyPool) removeInvalidFromDB() error { return fmt.Errorf("数据库未初始化") } - cacheMutex.Lock() - defer cacheMutex.Unlock() result, err := cacheDB.Exec(`DELETE FROM device_pool WHERE is_valid = 0`) if err != nil {