golang/hotime
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 85 Wiki Activity

feat(cache): 优化缓存系统并重构数据库连接管理

Browse Source 
- 将Redis连接方式改为连接池模式，提升连接复用效率
- 修复缓存注释错误，统一标识数据库缓存逻辑
- 添加数据检索结果非空验证，避免空指针异常
- 在数据库操作中添加读写锁保护，确保并发安全性
- 实现数据库查询和执行操作的重试机制，增强稳定性
- 更新配置文件中的缓存和数据库设置，优化缓存策略
- 重构README文档，补充框架特性和性能测试数据
- 添加示例路由配置，完善快速入门指南
This commit is contained in:
hoteas
2026-01-22 04:36:52 +08:00
parent 3455fb0a1c
commit b755519fc6
17 changed files with 2991 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/db.go
+100
View File
@@ -0,0 +1,100 @@
package db
import (
"code.hoteas.com/golang/hotime/cache"
. "code.hoteas.com/golang/hotime/common"
"database/sql"
_ "github.com/go-sql-driver/mysql"
_ "github.com/mattn/go-sqlite3"
"github.com/sirupsen/logrus"
"sync"
)
// HoTimeDB 数据库操作核心结构体
type HoTimeDB struct {
*sql.DB
ContextBase
DBName string
*cache.HoTimeCache
Log *logrus.Logger
Type string // 数据库类型: mysql, sqlite3, postgres
Prefix string
LastQuery string
LastData []interface{}
ConnectFunc func(err ...*Error) (*sql.DB, *sql.DB)
LastErr *Error
limit Slice
*sql.Tx //事务对象
SlaveDB *sql.DB // 从数据库
Mode int // mode为0生产模式,1为测试模式,2为开发模式
mu sync.RWMutex
limitMu sync.Mutex
Dialect Dialect // 数据库方言适配器
}
// SetConnect 设置数据库配置连接
func (that *HoTimeDB) SetConnect(connect func(err ...*Error) (master, slave *sql.DB), err ...*Error) {
that.ConnectFunc = connect
_ = that.InitDb(err...)
}
// InitDb 初始化数据库连接
func (that *HoTimeDB) InitDb(err ...*Error) *Error {
if len(err) != 0 {
that.LastErr = err[0]
}
that.DB, that.SlaveDB = that.ConnectFunc(that.LastErr)
if that.DB == nil {
return that.LastErr
}
e := that.DB.Ping()
that.LastErr.SetError(e)
if that.SlaveDB != nil {
e := that.SlaveDB.Ping()
that.LastErr.SetError(e)
}
// 根据数据库类型初始化方言适配器
if that.Dialect == nil {
that.initDialect()
}
return that.LastErr
}
// initDialect 根据数据库类型初始化方言
func (that *HoTimeDB) initDialect() {
switch that.Type {
case "postgres", "postgresql":
that.Dialect = &PostgreSQLDialect{}
case "sqlite3", "sqlite":
that.Dialect = &SQLiteDialect{}
default:
that.Dialect = &MySQLDialect{}
}
}
// GetDialect 获取当前方言适配器
func (that *HoTimeDB) GetDialect() Dialect {
if that.Dialect == nil {
that.initDialect()
}
return that.Dialect
}
// SetDialect 设置方言适配器
func (that *HoTimeDB) SetDialect(dialect Dialect) {
that.Dialect = dialect
}
// GetType 获取数据库类型
func (that *HoTimeDB) GetType() string {
return that.Type
}
// GetPrefix 获取表前缀
func (that *HoTimeDB) GetPrefix() string {
return that.Prefix
}
db/dialect.go
+249
View File
@@ -0,0 +1,249 @@
package db
import (
"fmt"
"strings"
)
// Dialect 数据库方言接口
// 用于处理不同数据库之间的语法差异
type Dialect interface {
// Quote 对表名/字段名添加引号
// MySQL 使用反引号 `name`
// PostgreSQL 使用双引号 "name"
// SQLite 使用双引号或方括号 "name" 或 [name]
Quote(name string) string
// Placeholder 生成占位符
// MySQL/SQLite 使用 ?
// PostgreSQL 使用 $1, $2, $3...
Placeholder(index int) string
// Placeholders 生成多个占位符，用逗号分隔
Placeholders(count int, startIndex int) string
// SupportsLastInsertId 是否支持 LastInsertId
// PostgreSQL 不支持，需要使用 RETURNING
SupportsLastInsertId() bool
// ReturningClause 生成 RETURNING 子句（用于 PostgreSQL）
ReturningClause(column string) string
// UpsertSQL 生成 Upsert 语句
// MySQL: INSERT ... ON DUPLICATE KEY UPDATE ...
// PostgreSQL: INSERT ... ON CONFLICT ... DO UPDATE SET ...
// SQLite: INSERT OR REPLACE / INSERT ... ON CONFLICT ...
UpsertSQL(table string, columns []string, uniqueKeys []string, updateColumns []string) string
// GetName 获取方言名称
GetName() string
}
// MySQLDialect MySQL 方言实现
type MySQLDialect struct{}
func (d *MySQLDialect) GetName() string {
return "mysql"
}
func (d *MySQLDialect) Quote(name string) string {
// 如果已经包含点号（表.字段）或空格（别名），不添加引号
if strings.Contains(name, ".") || strings.Contains(name, " ") {
return name
}
return "`" + name + "`"
}
func (d *MySQLDialect) Placeholder(index int) string {
return "?"
}
func (d *MySQLDialect) Placeholders(count int, startIndex int) string {
if count <= 0 {
return ""
}
placeholders := make([]string, count)
for i := 0; i < count; i++ {
placeholders[i] = "?"
}
return strings.Join(placeholders, ",")
}
func (d *MySQLDialect) SupportsLastInsertId() bool {
return true
}
func (d *MySQLDialect) ReturningClause(column string) string {
return "" // MySQL 不支持 RETURNING
}
func (d *MySQLDialect) UpsertSQL(table string, columns []string, uniqueKeys []string, updateColumns []string) string {
// INSERT INTO table (col1, col2) VALUES (?, ?)
// ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2)
quotedCols := make([]string, len(columns))
for i, col := range columns {
quotedCols[i] = d.Quote(col)
}
placeholders := d.Placeholders(len(columns), 1)
updateParts := make([]string, len(updateColumns))
for i, col := range updateColumns {
// 检查是否是 [#] 标记的直接 SQL
if strings.HasSuffix(col, "[#]") {
// 这种情况在调用处处理
updateParts[i] = col
} else {
quotedCol := d.Quote(col)
updateParts[i] = quotedCol + " = VALUES(" + quotedCol + ")"
}
}
return fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
d.Quote(table),
strings.Join(quotedCols, ", "),
placeholders,
strings.Join(updateParts, ", "))
}
// PostgreSQLDialect PostgreSQL 方言实现
type PostgreSQLDialect struct{}
func (d *PostgreSQLDialect) GetName() string {
return "postgres"
}
func (d *PostgreSQLDialect) Quote(name string) string {
// 如果已经包含点号（表.字段）或空格（别名），不添加引号
if strings.Contains(name, ".") || strings.Contains(name, " ") {
return name
}
return "\"" + name + "\""
}
func (d *PostgreSQLDialect) Placeholder(index int) string {
return fmt.Sprintf("$%d", index)
}
func (d *PostgreSQLDialect) Placeholders(count int, startIndex int) string {
if count <= 0 {
return ""
}
placeholders := make([]string, count)
for i := 0; i < count; i++ {
placeholders[i] = fmt.Sprintf("$%d", startIndex+i)
}
return strings.Join(placeholders, ",")
}
func (d *PostgreSQLDialect) SupportsLastInsertId() bool {
return false // PostgreSQL 需要使用 RETURNING
}
func (d *PostgreSQLDialect) ReturningClause(column string) string {
return " RETURNING " + d.Quote(column)
}
func (d *PostgreSQLDialect) UpsertSQL(table string, columns []string, uniqueKeys []string, updateColumns []string) string {
// INSERT INTO table (col1, col2) VALUES ($1, $2)
// ON CONFLICT (unique_key) DO UPDATE SET col1 = EXCLUDED.col1, col2 = EXCLUDED.col2
quotedCols := make([]string, len(columns))
for i, col := range columns {
quotedCols[i] = d.Quote(col)
}
placeholders := d.Placeholders(len(columns), 1)
quotedUniqueKeys := make([]string, len(uniqueKeys))
for i, key := range uniqueKeys {
quotedUniqueKeys[i] = d.Quote(key)
}
updateParts := make([]string, len(updateColumns))
for i, col := range updateColumns {
if strings.HasSuffix(col, "[#]") {
updateParts[i] = col
} else {
quotedCol := d.Quote(col)
updateParts[i] = quotedCol + " = EXCLUDED." + quotedCol
}
}
return fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO UPDATE SET %s",
d.Quote(table),
strings.Join(quotedCols, ", "),
placeholders,
strings.Join(quotedUniqueKeys, ", "),
strings.Join(updateParts, ", "))
}
// SQLiteDialect SQLite 方言实现
type SQLiteDialect struct{}
func (d *SQLiteDialect) GetName() string {
return "sqlite3"
}
func (d *SQLiteDialect) Quote(name string) string {
// 如果已经包含点号（表.字段）或空格（别名），不添加引号
if strings.Contains(name, ".") || strings.Contains(name, " ") {
return name
}
return "\"" + name + "\""
}
func (d *SQLiteDialect) Placeholder(index int) string {
return "?"
}
func (d *SQLiteDialect) Placeholders(count int, startIndex int) string {
if count <= 0 {
return ""
}
placeholders := make([]string, count)
for i := 0; i < count; i++ {
placeholders[i] = "?"
}
return strings.Join(placeholders, ",")
}
func (d *SQLiteDialect) SupportsLastInsertId() bool {
return true
}
func (d *SQLiteDialect) ReturningClause(column string) string {
return "" // SQLite 3.35+ 支持 RETURNING，但为兼容性暂不使用
}
func (d *SQLiteDialect) UpsertSQL(table string, columns []string, uniqueKeys []string, updateColumns []string) string {
// INSERT INTO table (col1, col2) VALUES (?, ?)
// ON CONFLICT (unique_key) DO UPDATE SET col1 = excluded.col1, col2 = excluded.col2
quotedCols := make([]string, len(columns))
for i, col := range columns {
quotedCols[i] = d.Quote(col)
}
placeholders := d.Placeholders(len(columns), 1)
quotedUniqueKeys := make([]string, len(uniqueKeys))
for i, key := range uniqueKeys {
quotedUniqueKeys[i] = d.Quote(key)
}
updateParts := make([]string, len(updateColumns))
for i, col := range updateColumns {
if strings.HasSuffix(col, "[#]") {
updateParts[i] = col
} else {
quotedCol := d.Quote(col)
updateParts[i] = quotedCol + " = excluded." + quotedCol
}
}
return fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO UPDATE SET %s",
d.Quote(table),
strings.Join(quotedCols, ", "),
placeholders,
strings.Join(quotedUniqueKeys, ", "),
strings.Join(updateParts, ", "))
}
db/hotimedb.go
+96 -46
View File
@@ -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
}