⚡️ perf(fqapi): 移除缓存层全局串行锁
- 删除 cache_manager/audio_proxy/device_key_pool 中 21 处 cacheMutex:database/sql 连接池本身并发安全,SQLite 已配 WAL + busy_timeout,PostgreSQL 无并发写限制 - contentRequestCount 三函数由互斥锁改为 atomic.Int64 - cache_manager 中 config.GetConfig() 改为 GetAppConfig(),消除配置未初始化时 log.Fatal 崩溃整个进程的隐患 - VACUUM 与写操作的低频并发冲突由 busy_timeout 兜底(已注释说明) - 新增 TestConcurrentCacheAccess:8 goroutine 并发读写缓存,-race 验证无死锁无竞争
This commit is contained in:
1 parent
24c7e4b6fd
commit
f0d89b462b
4 files changed
+68
-68
No files matched your search
@@ -35,8 +35,6 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri
|
|||||||
|
|
||||||
// 存储到数据库
|
// 存储到数据库
|
||||||
if cacheDB != nil {
|
if cacheDB != nil {
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
_, err := cacheDB.Exec(q(`
|
_, err := cacheDB.Exec(q(`
|
||||||
INSERT INTO audio_cache (key, original_url, device_id, headers, created_at)
|
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 headersJSON string
|
||||||
var createdAt int64
|
var createdAt int64
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
err := cacheDB.QueryRow(q(`
|
err := cacheDB.QueryRow(q(`
|
||||||
SELECT original_url, device_id, headers, created_at
|
SELECT original_url, device_id, headers, created_at
|
||||||
FROM audio_cache WHERE key = ?
|
FROM audio_cache WHERE key = ?
|
||||||
`), key).Scan(&originalURL, &deviceID, &headersJSON, &createdAt)
|
`), key).Scan(&originalURL, &deviceID, &headersJSON, &createdAt)
|
||||||
cacheMutex.Unlock()
|
|
||||||
|
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
return nil, "", fmt.Errorf("音频信息不存在,请先调用 /audio 接口")
|
return nil, "", fmt.Errorf("音频信息不存在,请先调用 /audio 接口")
|
||||||
@@ -88,9 +84,7 @@ func HandleAudioProxy(key string) ([]byte, string, error) {
|
|||||||
// 检查缓存是否过期(2小时,匹配CDN链接有效期)
|
// 检查缓存是否过期(2小时,匹配CDN链接有效期)
|
||||||
if time.Now().Unix()-createdAt > 7200 { // 2小时 = 7200秒
|
if time.Now().Unix()-createdAt > 7200 { // 2小时 = 7200秒
|
||||||
// 清理过期缓存
|
// 清理过期缓存
|
||||||
cacheMutex.Lock()
|
|
||||||
cacheDB.Exec(q("DELETE FROM audio_cache WHERE key = ?"), key)
|
cacheDB.Exec(q("DELETE FROM audio_cache WHERE key = ?"), key)
|
||||||
cacheMutex.Unlock()
|
|
||||||
return nil, "", fmt.Errorf("音频信息已过期,请重新调用 /audio 接口")
|
return nil, "", fmt.Errorf("音频信息已过期,请重新调用 /audio 接口")
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -121,9 +115,6 @@ func CleanExpiredAudioCache() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
// 删除2小时前的缓存
|
// 删除2小时前的缓存
|
||||||
result, err := cacheDB.Exec(q("DELETE FROM audio_cache WHERE created_at < ?"), time.Now().Unix()-7200)
|
result, err := cacheDB.Exec(q("DELETE FROM audio_cache WHERE created_at < ?"), time.Now().Unix()-7200)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
_ "github.com/jackc/pgx/v5/stdlib"
|
_ "github.com/jackc/pgx/v5/stdlib"
|
||||||
@@ -22,10 +22,8 @@ var (
|
|||||||
cacheDB *sql.DB
|
cacheDB *sql.DB
|
||||||
dbDriver string // "sqlite3" | "pgx"
|
dbDriver string // "sqlite3" | "pgx"
|
||||||
redisCache *redis.Client
|
redisCache *redis.Client
|
||||||
cacheMutex sync.Mutex
|
|
||||||
cacheLogger = common.NewLogger("cache_manager", "INFO", "")
|
cacheLogger = common.NewLogger("cache_manager", "INFO", "")
|
||||||
contentRequestCount int // content接口访问计数
|
contentRequestCount atomic.Int64 // content接口访问计数
|
||||||
countMutex sync.Mutex
|
|
||||||
cacheCleanupTicker *time.Ticker // 缓存清理定时器
|
cacheCleanupTicker *time.Ticker // 缓存清理定时器
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -49,7 +47,7 @@ func q(query string) string {
|
|||||||
|
|
||||||
// InitRedis 初始化 Redis 缓存(启用后正文缓存与音频代理缓存走 Redis)
|
// InitRedis 初始化 Redis 缓存(启用后正文缓存与音频代理缓存走 Redis)
|
||||||
func InitRedis() error {
|
func InitRedis() error {
|
||||||
cfg := config.GetConfig()
|
cfg := config.GetAppConfig()
|
||||||
if cfg == nil || !cfg.Cache.Redis.Enabled {
|
if cfg == nil || !cfg.Cache.Redis.Enabled {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -90,7 +88,7 @@ func InitDB() error {
|
|||||||
return nil // 已初始化
|
return nil // 已初始化
|
||||||
}
|
}
|
||||||
|
|
||||||
cfg := config.GetConfig()
|
cfg := config.GetAppConfig()
|
||||||
backend := "sqlite"
|
backend := "sqlite"
|
||||||
dsn := "./cache.db"
|
dsn := "./cache.db"
|
||||||
if cfg != nil {
|
if cfg != nil {
|
||||||
@@ -250,7 +248,7 @@ func createTables() error {
|
|||||||
|
|
||||||
// isCacheEnabled 检查正文内容缓存是否启用
|
// isCacheEnabled 检查正文内容缓存是否启用
|
||||||
func isCacheEnabled() bool {
|
func isCacheEnabled() bool {
|
||||||
cfg := config.GetConfig()
|
cfg := config.GetAppConfig()
|
||||||
if cfg == nil {
|
if cfg == nil {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
@@ -271,9 +269,6 @@ func GetCache(key string) string {
|
|||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
var value string
|
var value string
|
||||||
err := cacheDB.QueryRow(q("SELECT value FROM cache WHERE key = ?"), key).Scan(&value)
|
err := cacheDB.QueryRow(q("SELECT value FROM cache WHERE key = ?"), key).Scan(&value)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -316,9 +311,6 @@ func UpdateCache(key string, value string, expi int) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
cacheData := CacheData{
|
cacheData := CacheData{
|
||||||
Data: value,
|
Data: value,
|
||||||
Expi: float64(time.Now().Unix() + int64(expi)),
|
Expi: float64(time.Now().Unix() + int64(expi)),
|
||||||
@@ -341,7 +333,6 @@ func UpdateCache(key string, value string, expi int) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
// CloseCache 关闭缓存数据库
|
// CloseCache 关闭缓存数据库
|
||||||
func CloseCache() {
|
func CloseCache() {
|
||||||
if cacheDB != nil {
|
if cacheDB != nil {
|
||||||
@@ -349,18 +340,12 @@ func CloseCache() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// CleanupAllCache 立即清理所有正文缓存
|
// CleanupAllCache 立即清理所有正文缓存
|
||||||
func CleanupAllCache() error {
|
func CleanupAllCache() error {
|
||||||
if cacheDB == nil {
|
if cacheDB == nil {
|
||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
// 删除所有缓存数据
|
// 删除所有缓存数据
|
||||||
_, err := cacheDB.Exec("DELETE FROM cache")
|
_, err := cacheDB.Exec("DELETE FROM cache")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -401,7 +386,7 @@ func CleanupAllCache() error {
|
|||||||
|
|
||||||
// CleanupExpiredCache 清理过期的正文缓存
|
// CleanupExpiredCache 清理过期的正文缓存
|
||||||
func CleanupExpiredCache() error {
|
func CleanupExpiredCache() error {
|
||||||
cfg := config.GetConfig()
|
cfg := config.GetAppConfig()
|
||||||
if cfg == nil {
|
if cfg == nil {
|
||||||
return fmt.Errorf("配置未加载")
|
return fmt.Errorf("配置未加载")
|
||||||
}
|
}
|
||||||
@@ -420,9 +405,6 @@ func CleanupExpiredCache() error {
|
|||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
// 计算过期时间点(当前时间 - cleanup_seconds)
|
// 计算过期时间点(当前时间 - cleanup_seconds)
|
||||||
expiredTime := time.Now().Unix() - int64(cfg.Cache.CleanupSeconds)
|
expiredTime := time.Now().Unix() - int64(cfg.Cache.CleanupSeconds)
|
||||||
|
|
||||||
@@ -446,7 +428,7 @@ func CleanupExpiredCache() error {
|
|||||||
|
|
||||||
// StartCacheCleanup 启动缓存自动清理任务
|
// StartCacheCleanup 启动缓存自动清理任务
|
||||||
func StartCacheCleanup() {
|
func StartCacheCleanup() {
|
||||||
cfg := config.GetConfig()
|
cfg := config.GetAppConfig()
|
||||||
if cfg == nil || !cfg.Cache.CleanupEnabled || cfg.Cache.CleanupSeconds <= 0 {
|
if cfg == nil || !cfg.Cache.CleanupEnabled || cfg.Cache.CleanupSeconds <= 0 {
|
||||||
cacheLogger.Info("缓存自动清理未启用或配置无效,跳过启动")
|
cacheLogger.Info("缓存自动清理未启用或配置无效,跳过启动")
|
||||||
return
|
return
|
||||||
@@ -488,36 +470,21 @@ func StopCacheCleanup() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// IncrementContentRequestCount 增加content接口访问计数
|
// IncrementContentRequestCount 增加content接口访问计数
|
||||||
func IncrementContentRequestCount() int {
|
func IncrementContentRequestCount() int { return int(contentRequestCount.Add(1)) }
|
||||||
countMutex.Lock()
|
|
||||||
defer countMutex.Unlock()
|
|
||||||
contentRequestCount++
|
|
||||||
return contentRequestCount
|
|
||||||
}
|
|
||||||
|
|
||||||
// ResetContentRequestCount 重置访问计数
|
// ResetContentRequestCount 重置访问计数
|
||||||
func ResetContentRequestCount() {
|
func ResetContentRequestCount() { contentRequestCount.Store(0) }
|
||||||
countMutex.Lock()
|
|
||||||
defer countMutex.Unlock()
|
|
||||||
contentRequestCount = 0
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetContentRequestCount 获取当前访问计数
|
// GetContentRequestCount 获取当前访问计数
|
||||||
func GetContentRequestCount() int {
|
func GetContentRequestCount() int { return int(contentRequestCount.Load()) }
|
||||||
countMutex.Lock()
|
|
||||||
defer countMutex.Unlock()
|
|
||||||
return contentRequestCount
|
|
||||||
}
|
|
||||||
|
|
||||||
// VacuumDatabase 压缩数据库,释放磁盘空间
|
// VacuumDatabase 压缩数据库,释放磁盘空间
|
||||||
|
// ponytail: 删了全局串行锁,VACUUM 与写操作的并发冲突由 busy_timeout(5s) 兜底,失败仅记日志
|
||||||
func VacuumDatabase() error {
|
func VacuumDatabase() error {
|
||||||
if cacheDB == nil {
|
if cacheDB == nil {
|
||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
cacheLogger.Info("开始执行数据库压缩...")
|
cacheLogger.Info("开始执行数据库压缩...")
|
||||||
if dbDriver == "pgx" {
|
if dbDriver == "pgx" {
|
||||||
cacheLogger.Info("PostgreSQL 由 autovacuum 自动管理,无需手动压缩")
|
cacheLogger.Info("PostgreSQL 由 autovacuum 自动管理,无需手动压缩")
|
||||||
@@ -539,9 +506,6 @@ func GetDatabaseStats() (map[string]interface{}, error) {
|
|||||||
return nil, fmt.Errorf("数据库未初始化")
|
return nil, fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
stats := make(map[string]interface{})
|
stats := make(map[string]interface{})
|
||||||
|
|
||||||
// 获取cache表记录数
|
// 获取cache表记录数
|
||||||
@@ -596,9 +560,6 @@ func SaveStatistics(stats map[string]int64) error {
|
|||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
_, err := cacheDB.Exec(q(`
|
_, err := cacheDB.Exec(q(`
|
||||||
UPDATE statistics
|
UPDATE statistics
|
||||||
SET total_requests = ?,
|
SET total_requests = ?,
|
||||||
@@ -630,9 +591,6 @@ func LoadStatistics() (map[string]int64, error) {
|
|||||||
return nil, fmt.Errorf("数据库未初始化")
|
return nil, fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
var total, search, detail, catalog, content, manga, audio, video, category int64
|
var total, search, detail, catalog, content, manga, audio, video, category int64
|
||||||
|
|
||||||
err := cacheDB.QueryRow(`
|
err := cacheDB.QueryRow(`
|
||||||
|
|||||||
@@ -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()
|
||||||
|
}
|
||||||
@@ -399,8 +399,6 @@ func (p *DeviceKeyPool) saveDeviceToDB(item *DeviceKeyItem) error {
|
|||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
// RETURNING id 同时兼容 SQLite(3.35+) 与 PostgreSQL,替代 LastInsertId()
|
// RETURNING id 同时兼容 SQLite(3.35+) 与 PostgreSQL,替代 LastInsertId()
|
||||||
err := cacheDB.QueryRow(q(`
|
err := cacheDB.QueryRow(q(`
|
||||||
@@ -419,8 +417,6 @@ func (p *DeviceKeyPool) updateDeviceInDB(item *DeviceKeyItem) error {
|
|||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
_, err := cacheDB.Exec(q(`
|
_, err := cacheDB.Exec(q(`
|
||||||
UPDATE device_pool
|
UPDATE device_pool
|
||||||
@@ -438,8 +434,6 @@ func (p *DeviceKeyPool) removeInvalidFromDB() error {
|
|||||||
return fmt.Errorf("数据库未初始化")
|
return fmt.Errorf("数据库未初始化")
|
||||||
}
|
}
|
||||||
|
|
||||||
cacheMutex.Lock()
|
|
||||||
defer cacheMutex.Unlock()
|
|
||||||
|
|
||||||
result, err := cacheDB.Exec(`DELETE FROM device_pool WHERE is_valid = 0`)
|
result, err := cacheDB.Exec(`DELETE FROM device_pool WHERE is_valid = 0`)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in new issue
Block a user