feat(cache): 优化缓存系统并重构数据库连接管理
- 将Redis连接方式改为连接池模式，提升连接复用效率 - 修复缓存注释错误，统一标识数据库缓存逻辑 - 添加数据检索结果非空验证，避免空指针异常 - 在数据库操作中添加读写锁保护，确保并发安全性 - 实现数据库查询和执行操作的重试机制，增强稳定性 - 更新配置文件中的缓存和数据库设置，优化缓存策略 - 重构README文档，补充框架特性和性能测试数据 - 添加示例路由配置，完善快速入门指南
This commit is contained in:
+96
-46
@@ -13,6 +13,7 @@ import (
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type HoTimeDB struct {
|
||||
@@ -31,6 +32,8 @@ type HoTimeDB struct {
|
||||
*sql.Tx //事务对象
|
||||
SlaveDB *sql.DB
|
||||
Mode int //mode为0生产模式,1、为测试模式、2为开发模式
|
||||
mu sync.RWMutex
|
||||
limitMu sync.Mutex
|
||||
}
|
||||
|
||||
type HotimeDBBuilder struct {
|
||||
@@ -271,11 +274,25 @@ func (that *HoTimeDB) GetType() string {
|
||||
|
||||
// Action 事务，如果action返回true则执行成功；false则回滚
|
||||
func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSuccess bool) {
|
||||
db := HoTimeDB{that.DB, that.ContextBase, that.DBName,
|
||||
that.HoTimeCache, that.Log, that.Type,
|
||||
that.Prefix, that.LastQuery, that.LastData,
|
||||
that.ConnectFunc, that.LastErr, that.limit, that.Tx,
|
||||
that.SlaveDB, that.Mode}
|
||||
db := HoTimeDB{
|
||||
DB: that.DB,
|
||||
ContextBase: that.ContextBase,
|
||||
DBName: that.DBName,
|
||||
HoTimeCache: that.HoTimeCache,
|
||||
Log: that.Log,
|
||||
Type: that.Type,
|
||||
Prefix: that.Prefix,
|
||||
LastQuery: that.LastQuery,
|
||||
LastData: that.LastData,
|
||||
ConnectFunc: that.ConnectFunc,
|
||||
LastErr: that.LastErr,
|
||||
limit: that.limit,
|
||||
Tx: that.Tx,
|
||||
SlaveDB: that.SlaveDB,
|
||||
Mode: that.Mode,
|
||||
mu: sync.RWMutex{},
|
||||
limitMu: sync.Mutex{},
|
||||
}
|
||||
|
||||
//tx, err := db.BeginTx(context.Background(), &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||
tx, err := db.Begin()
|
||||
@@ -331,23 +348,33 @@ func (that *HoTimeDB) Page(page, pageRow int) *HoTimeDB {
|
||||
page = 1
|
||||
}
|
||||
|
||||
that.limitMu.Lock()
|
||||
that.limit = Slice{page, pageRow}
|
||||
that.limitMu.Unlock()
|
||||
return that
|
||||
}
|
||||
|
||||
func (that *HoTimeDB) PageSelect(table string, qu ...interface{}) []Map {
|
||||
that.limitMu.Lock()
|
||||
limit := that.limit
|
||||
that.limit = nil // 使用后清空，避免影响下次调用
|
||||
that.limitMu.Unlock()
|
||||
|
||||
if limit == nil {
|
||||
return that.Select(table, qu...)
|
||||
}
|
||||
|
||||
if len(qu) == 1 {
|
||||
qu = append(qu, Map{"LIMIT": that.limit})
|
||||
qu = append(qu, Map{"LIMIT": limit})
|
||||
}
|
||||
if len(qu) == 2 {
|
||||
temp := DeepCopyMap(qu[1]).(Map)
|
||||
temp["LIMIT"] = that.limit
|
||||
temp["LIMIT"] = limit
|
||||
qu[1] = temp
|
||||
}
|
||||
if len(qu) == 3 {
|
||||
temp := DeepCopyMap(qu[2]).(Map)
|
||||
temp["LIMIT"] = that.limit
|
||||
temp["LIMIT"] = limit
|
||||
qu[2] = temp
|
||||
}
|
||||
//fmt.Println(qu)
|
||||
@@ -540,17 +567,28 @@ func (that *HoTimeDB) md5(query string, args ...interface{}) string {
|
||||
}
|
||||
|
||||
func (that *HoTimeDB) Query(query string, args ...interface{}) []Map {
|
||||
return that.queryWithRetry(query, false, args...)
|
||||
}
|
||||
|
||||
// queryWithRetry 内部查询方法，支持重试标记
|
||||
func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interface{}) []Map {
|
||||
// 保存调试信息（加锁保护）
|
||||
that.mu.Lock()
|
||||
that.LastQuery = query
|
||||
that.LastData = args
|
||||
that.mu.Unlock()
|
||||
|
||||
defer func() {
|
||||
if that.Mode != 0 {
|
||||
that.mu.RLock()
|
||||
that.Log.Info("SQL:"+that.LastQuery, " DATA:", that.LastData, " ERROR:", that.LastErr.GetError())
|
||||
that.mu.RUnlock()
|
||||
}
|
||||
}()
|
||||
//fmt.Println(query)
|
||||
|
||||
var err error
|
||||
var resl *sql.Rows
|
||||
|
||||
that.LastQuery = query
|
||||
that.LastData = args
|
||||
//主从数据库切换，只有select语句有从数据库
|
||||
db := that.DB
|
||||
if that.SlaveDB != nil {
|
||||
@@ -562,8 +600,15 @@ func (that *HoTimeDB) Query(query string, args ...interface{}) []Map {
|
||||
that.LastErr.SetError(err)
|
||||
return nil
|
||||
}
|
||||
for key, _ := range args {
|
||||
arg := args[key]
|
||||
|
||||
// 处理参数中的 slice 类型
|
||||
processedArgs := make([]interface{}, len(args))
|
||||
copy(processedArgs, args)
|
||||
for key := range processedArgs {
|
||||
arg := processedArgs[key]
|
||||
if arg == nil {
|
||||
continue
|
||||
}
|
||||
argType := reflect.ValueOf(arg).Type().String()
|
||||
if strings.Contains(argType, "[]") || strings.Contains(argType, "Slice") {
|
||||
argLis := ObjToSlice(arg)
|
||||
@@ -576,29 +621,24 @@ func (that *HoTimeDB) Query(query string, args ...interface{}) []Map {
|
||||
argStr += ObjToStr(argLis[i]) + ","
|
||||
}
|
||||
}
|
||||
args[key] = argStr
|
||||
processedArgs[key] = argStr
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if that.Tx != nil {
|
||||
resl, err = that.Tx.Query(query, args...)
|
||||
resl, err = that.Tx.Query(query, processedArgs...)
|
||||
} else {
|
||||
resl, err = db.Query(query, args...)
|
||||
}
|
||||
|
||||
if err != nil && that.LastErr.GetError() != nil &&
|
||||
that.LastErr.GetError().Error() == err.Error() {
|
||||
return nil
|
||||
resl, err = db.Query(query, processedArgs...)
|
||||
}
|
||||
|
||||
that.LastErr.SetError(err)
|
||||
if err != nil {
|
||||
|
||||
if err = db.Ping(); err == nil {
|
||||
return that.Query(query, args...)
|
||||
// 如果还没重试过，尝试 Ping 后重试一次
|
||||
if !retried {
|
||||
if pingErr := db.Ping(); pingErr == nil {
|
||||
return that.queryWithRetry(query, true, args...)
|
||||
}
|
||||
}
|
||||
that.LastErr.SetError(err)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -606,13 +646,25 @@ func (that *HoTimeDB) Query(query string, args ...interface{}) []Map {
|
||||
}
|
||||
|
||||
func (that *HoTimeDB) Exec(query string, args ...interface{}) (sql.Result, *Error) {
|
||||
defer func() {
|
||||
if that.Mode != 0 {
|
||||
that.Log.Info("SQL: "+that.LastQuery, " DATA: ", that.LastData, " ERROR: ", that.LastErr.GetError())
|
||||
}
|
||||
}()
|
||||
return that.execWithRetry(query, false, args...)
|
||||
}
|
||||
|
||||
// execWithRetry 内部执行方法，支持重试标记
|
||||
func (that *HoTimeDB) execWithRetry(query string, retried bool, args ...interface{}) (sql.Result, *Error) {
|
||||
// 保存调试信息（加锁保护）
|
||||
that.mu.Lock()
|
||||
that.LastQuery = query
|
||||
that.LastData = args
|
||||
that.mu.Unlock()
|
||||
|
||||
defer func() {
|
||||
if that.Mode != 0 {
|
||||
that.mu.RLock()
|
||||
that.Log.Info("SQL: "+that.LastQuery, " DATA: ", that.LastData, " ERROR: ", that.LastErr.GetError())
|
||||
that.mu.RUnlock()
|
||||
}
|
||||
}()
|
||||
|
||||
var e error
|
||||
var resl sql.Result
|
||||
|
||||
@@ -622,8 +674,11 @@ func (that *HoTimeDB) Exec(query string, args ...interface{}) (sql.Result, *Erro
|
||||
return nil, that.LastErr
|
||||
}
|
||||
|
||||
for key, _ := range args {
|
||||
arg := args[key]
|
||||
// 处理参数中的 slice 类型
|
||||
processedArgs := make([]interface{}, len(args))
|
||||
copy(processedArgs, args)
|
||||
for key := range processedArgs {
|
||||
arg := processedArgs[key]
|
||||
argType := ""
|
||||
if arg != nil {
|
||||
argType = reflect.ValueOf(arg).Type().String()
|
||||
@@ -640,29 +695,24 @@ func (that *HoTimeDB) Exec(query string, args ...interface{}) (sql.Result, *Erro
|
||||
argStr += ObjToStr(argLis[i]) + ","
|
||||
}
|
||||
}
|
||||
args[key] = argStr
|
||||
processedArgs[key] = argStr
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if that.Tx != nil {
|
||||
resl, e = that.Tx.Exec(query, args...)
|
||||
resl, e = that.Tx.Exec(query, processedArgs...)
|
||||
} else {
|
||||
resl, e = that.DB.Exec(query, args...)
|
||||
resl, e = that.DB.Exec(query, processedArgs...)
|
||||
}
|
||||
|
||||
if e != nil && that.LastErr.GetError() != nil &&
|
||||
that.LastErr.GetError().Error() == e.Error() {
|
||||
return resl, that.LastErr
|
||||
}
|
||||
that.LastErr.SetError(e)
|
||||
//判断是否连接断开了
|
||||
//判断是否连接断开了，如果还没重试过，尝试重试一次
|
||||
if e != nil {
|
||||
|
||||
if e = that.DB.Ping(); e == nil {
|
||||
return that.Exec(query, args...)
|
||||
if !retried {
|
||||
if pingErr := that.DB.Ping(); pingErr == nil {
|
||||
return that.execWithRetry(query, true, args...)
|
||||
}
|
||||
}
|
||||
that.LastErr.SetError(e)
|
||||
return resl, that.LastErr
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user