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

refactor(logging): 迁移日志记录到 Zerolog 并优化错误处理

Browse Source 
- 将日志记录库从 Logrus 替换为 Zerolog，提升性能和灵活性
- 更新各个模块的日志记录方式，确保一致性
- 优化错误处理逻辑，确保在发生错误时能够正确记录并传递错误信息
- 移除不再使用的错误处理字段，简化代码结构
- 更新相关文档以反映新的日志记录和错误处理机制
This commit is contained in:
hoteas
2026-04-13 00:38:50 +08:00
parent 86daa3153e
commit 9a9b9c83ff
445 changed files with 59349 additions and 13297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/crud.go
+21 -13
View File
@@ -8,6 +8,10 @@ import (
. "code.hoteas.com/golang/hotime/common"
)
func (that *HoTimeDB) isCacheTable(table string) bool {
return table == "cached" || strings.HasPrefix(table, "hotime_cache")
}
// Page 设置分页参数
// page: 页码（从1开始）
// pageRow: 每页数量
@@ -135,7 +139,7 @@ func (that *HoTimeDB) Select(table string, qu ...interface{}) []Map {
qs = append(qs, resWhere...)
md5 := that.md5(query, qs...)
if that.testTx == nil && that.HoTimeCache != nil && table != "cached" {
if that.testTx == nil && that.HoTimeCache != nil && !that.isCacheTable(table) {
cacheData := that.HoTimeCache.Db(table + ":" + md5)
if cacheData != nil && cacheData.Data != nil {
return cacheData.ToMapArray()
@@ -148,7 +152,7 @@ func (that *HoTimeDB) Select(table string, qu ...interface{}) []Map {
res = []Map{}
}
if that.testTx == nil && that.HoTimeCache != nil && table != "cached" {
if that.testTx == nil && that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+":"+md5, res)
}
@@ -295,16 +299,20 @@ func (that *HoTimeDB) Insert(table string, data map[string]interface{}) int64 {
}
} else {
res, err := that.Exec(query, values...)
if err.GetError() == nil && res != nil {
if err == nil && res != nil {
id1, e := res.LastInsertId()
that.LastErr.SetError(e)
if e != nil {
dbErr := &DBError{}
dbErr.SetError(e)
that.lastError.Store(dbErr)
}
id = id1
}
}
// 如果插入成功，删除缓存
if id != 0 {
if that.HoTimeCache != nil && table != "cached" {
if that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+"*", nil)
}
}
@@ -386,13 +394,13 @@ func (that *HoTimeDB) Inserts(table string, dataList []Map) int64 {
res, err := that.Exec(query, values...)
rows64 := int64(0)
if err.GetError() == nil && res != nil {
if err == nil && res != nil {
rows64, _ = res.RowsAffected()
}
// 如果插入成功，删除缓存
if rows64 != 0 {
if that.HoTimeCache != nil && table != "cached" {
if that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+"*", nil)
}
}
@@ -492,13 +500,13 @@ func (that *HoTimeDB) Upsert(table string, data Map, uniqueKeys Slice, updateCol
res, err := that.Exec(query, values...)
rows := int64(0)
if err.GetError() == nil && res != nil {
if err == nil && res != nil {
rows, _ = res.RowsAffected()
}
// 清除缓存
if rows != 0 {
if that.HoTimeCache != nil && table != "cached" {
if that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+"*", nil)
}
}
@@ -693,13 +701,13 @@ func (that *HoTimeDB) Update(table string, data Map, where Map) int64 {
res, err := that.Exec(query, qs...)
rows := int64(0)
if err.GetError() == nil && res != nil {
if err == nil && res != nil {
rows, _ = res.RowsAffected()
}
// 如果更新成功，则删除缓存
if rows != 0 {
if that.HoTimeCache != nil && table != "cached" {
if that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+"*", nil)
}
}
@@ -717,13 +725,13 @@ func (that *HoTimeDB) Delete(table string, data map[string]interface{}) int64 {
res, err := that.Exec(query, resWhere...)
rows := int64(0)
if err.GetError() == nil && res != nil {
if err == nil && res != nil {
rows, _ = res.RowsAffected()
}
// 如果删除成功，删除对应缓存
if rows != 0 {
if that.HoTimeCache != nil && table != "cached" {
if that.HoTimeCache != nil && !that.isCacheTable(table) {
_ = that.HoTimeCache.Db(table+"*", nil)
}
}
db/db.go
+75 -34
View File
@@ -4,73 +4,114 @@ import (
"bytes"
"code.hoteas.com/golang/hotime/cache"
. "code.hoteas.com/golang/hotime/common"
"code.hoteas.com/golang/hotime/log"
"database/sql"
"strings"
"sync"
"sync/atomic"
_ "gitee.com/chunanyong/dm"
_ "github.com/go-sql-driver/mysql"
_ "github.com/mattn/go-sqlite3"
"github.com/sirupsen/logrus"
)
// DBError 最后一次 SQL 操作的结构化快照
// 嵌入框架 Error 类型，与 Obj 等保持一致的错误 API（HasError/GetError/Unwrap）
type DBError struct {
Error
Query string
Data []interface{}
}
// 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 // 数据库方言适配器
testTx *sql.Tx // 测试事务：设置后所有操作都在此事务内，测试结束回滚
testMu *sync.Mutex // 保护 testTx 单连接不被并发访问
testLogBuf *bytes.Buffer // 测试模式：SQL 日志写入此 buffer 而非直接打印，失败时才输出
testLogBufMu sync.Mutex // 保护 testLogBuf 并发写入
txFailed *bool // 事务内是否有SQL出错，出错则事务必须回滚（指针确保按值传递 HoTimeDB 时状态共享）
Log *log.Logger
Type string // 数据库类型: mysql, sqlite3, postgres
Prefix string
ConnectFunc func() (*sql.DB, *sql.DB)
lastError atomic.Pointer[DBError] // 无锁读写，始终记录最后一次 SQL 操作
limit Slice
*sql.Tx //事务对象
SlaveDB *sql.DB // 从数据库
limitMu sync.Mutex
Dialect Dialect // 数据库方言适配器
testTx *sql.Tx // 测试事务：设置后所有操作都在此事务内，测试结束回滚
testMu *sync.Mutex // 保护 testTx 单连接不被并发访问
testLogBuf *bytes.Buffer // 测试模式：SQL 日志写入此 buffer 而非直接打印，失败时才输出
testLogBufMu sync.Mutex // 保护 testLogBuf 并发写入
txFailed *bool // 事务内是否有SQL出错，出错则事务必须回滚（指针确保按值传递 HoTimeDB 时状态共享）
}
// GetLast 返回最后一次 SQL 操作的完整快照（Query + Data + Error）
func (that *HoTimeDB) GetLast() *DBError {
return that.lastError.Load()
}
// GetLastError 返回最后一次 SQL 的错误（nil = 成功）
func (that *HoTimeDB) GetLastError() error {
if e := that.lastError.Load(); e != nil {
return e.GetError()
}
return nil
}
// GetLastQuery 返回最后一次执行的 SQL 语句
func (that *HoTimeDB) GetLastQuery() string {
if e := that.lastError.Load(); e != nil {
return e.Query
}
return ""
}
// GetLastData 返回最后一次 SQL 的参数
func (that *HoTimeDB) GetLastData() []interface{} {
if e := that.lastError.Load(); e != nil {
return e.Data
}
return nil
}
// SetConnect 设置数据库配置连接
func (that *HoTimeDB) SetConnect(connect func(err ...*Error) (master, slave *sql.DB), err ...*Error) {
func (that *HoTimeDB) SetConnect(connect func() (master, slave *sql.DB)) {
that.ConnectFunc = connect
_ = that.InitDb(err...)
that.InitDb()
}
// InitDb 初始化数据库连接
func (that *HoTimeDB) InitDb(err ...*Error) *Error {
if len(err) != 0 {
that.LastErr = err[0]
}
that.DB, that.SlaveDB = that.ConnectFunc(that.LastErr)
func (that *HoTimeDB) InitDb() {
that.DB, that.SlaveDB = that.ConnectFunc()
if that.DB == nil {
return that.LastErr
return
}
e := that.DB.Ping()
that.LastErr.SetError(e)
if e != nil {
dbErr := &DBError{}
dbErr.SetError(e)
that.lastError.Store(dbErr)
if that.Log != nil {
that.Log.Error().Err(e).Msg("数据库连接失败")
}
}
if that.SlaveDB != nil {
e := that.SlaveDB.Ping()
that.LastErr.SetError(e)
if e != nil {
dbErr := &DBError{}
dbErr.SetError(e)
that.lastError.Store(dbErr)
if that.Log != nil {
that.Log.Error().Err(e).Msg("从数据库连接失败")
}
}
}
// 根据数据库类型初始化方言适配器
if that.Dialect == nil {
that.initDialect()
}
return that.LastErr
}
// initDialect 根据数据库类型初始化方言
db/query.go
+65 -73
View File
@@ -28,55 +28,37 @@ func (that *HoTimeDB) Query(query string, args ...interface{}) []Map {
// queryWithRetry 内部查询方法，支持重试标记
func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interface{}) []Map {
// 预处理数组占位符 ?[]
query, args = that.expandArrayPlaceholder(query, args)
// 保存调试信息（加锁保护）
that.mu.Lock()
that.LastQuery = query
that.LastData = args
that.mu.Unlock()
var sqlErr error
defer func() {
if that.Mode != 0 {
that.mu.RLock()
msg := fmt.Sprintf("SQL:%s DATA:%v ERROR:%v", that.LastQuery, that.LastData, that.LastErr.GetError())
that.mu.RUnlock()
that.testLogBufMu.Lock()
if that.testLogBuf != nil {
that.testLogBuf.WriteString(msg + "\n")
} else {
that.Log.Info(msg)
}
that.testLogBufMu.Unlock()
}
e := &DBError{Query: query, Data: args}
e.SetError(sqlErr)
that.lastError.Store(e)
that.logSQL(query, args, sqlErr)
}()
var err error
var resl *sql.Rows
// 主从数据库切换，只有select语句有从数据库
db := that.DB
if that.SlaveDB != nil {
db = that.SlaveDB
}
if db == nil {
err = errors.New("没有初始化数据库")
that.LastErr.SetError(err)
sqlErr = errors.New("没有初始化数据库")
return nil
}
// 处理参数中的 slice 类型
processedArgs := that.processArgs(args)
if that.testTx != nil {
if that.testMu != nil {
that.testMu.Lock()
}
resl, err = that.testTx.Query(query, processedArgs...)
that.LastErr.SetError(err)
if err != nil {
resl, sqlErr = that.testTx.Query(query, processedArgs...)
if sqlErr != nil {
if that.txFailed != nil {
*that.txFailed = true
}
@@ -91,13 +73,12 @@ func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interfa
}
return result
} else if that.Tx != nil {
resl, err = that.Tx.Query(query, processedArgs...)
resl, sqlErr = that.Tx.Query(query, processedArgs...)
} else {
resl, err = db.Query(query, processedArgs...)
resl, sqlErr = db.Query(query, processedArgs...)
}
that.LastErr.SetError(err)
if err != nil {
if sqlErr != nil {
if that.Tx != nil {
if that.txFailed != nil {
*that.txFailed = true
@@ -105,6 +86,7 @@ func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interfa
}
if !retried && that.Tx == nil {
if pingErr := db.Ping(); pingErr == nil {
sqlErr = nil // 重置，让 retry 的 defer 覆盖
return that.queryWithRetry(query, true, args...)
}
}
@@ -115,88 +97,94 @@ func (that *HoTimeDB) queryWithRetry(query string, retried bool, args ...interfa
}
// Exec 执行非查询 SQL
func (that *HoTimeDB) Exec(query string, args ...interface{}) (sql.Result, *Error) {
func (that *HoTimeDB) Exec(query string, args ...interface{}) (sql.Result, error) {
return that.execWithRetry(query, false, args...)
}
// execWithRetry 内部执行方法，支持重试标记
func (that *HoTimeDB) execWithRetry(query string, retried bool, args ...interface{}) (sql.Result, *Error) {
// 预处理数组占位符 ?[]
func (that *HoTimeDB) execWithRetry(query string, retried bool, args ...interface{}) (sql.Result, error) {
query, args = that.expandArrayPlaceholder(query, args)
// 保存调试信息（加锁保护）
that.mu.Lock()
that.LastQuery = query
that.LastData = args
that.mu.Unlock()
defer func() {
if that.Mode != 0 {
that.mu.RLock()
msg := fmt.Sprintf("SQL:%s DATA:%v ERROR:%v", that.LastQuery, that.LastData, that.LastErr.GetError())
that.mu.RUnlock()
that.testLogBufMu.Lock()
if that.testLogBuf != nil {
that.testLogBuf.WriteString(msg + "\n")
} else {
that.Log.Info(msg)
}
that.testLogBufMu.Unlock()
}
}()
var e error
var sqlErr error
var resl sql.Result
defer func() {
e := &DBError{Query: query, Data: args}
e.SetError(sqlErr)
that.lastError.Store(e)
that.logSQL(query, args, sqlErr)
}()
if that.DB == nil {
err := errors.New("没有初始化数据库")
that.LastErr.SetError(err)
return nil, that.LastErr
sqlErr = errors.New("没有初始化数据库")
return nil, sqlErr
}
// 处理参数中的 slice 类型
processedArgs := that.processArgs(args)
if that.testTx != nil {
if that.testMu != nil {
that.testMu.Lock()
}
resl, e = that.testTx.Exec(query, processedArgs...)
resl, sqlErr = that.testTx.Exec(query, processedArgs...)
if that.testMu != nil {
that.testMu.Unlock()
}
} else if that.Tx != nil {
resl, e = that.Tx.Exec(query, processedArgs...)
resl, sqlErr = that.Tx.Exec(query, processedArgs...)
} else {
resl, e = that.DB.Exec(query, processedArgs...)
resl, sqlErr = that.DB.Exec(query, processedArgs...)
}
that.LastErr.SetError(e)
if e != nil {
// testTx 模式下不走连接池重试，避免 busy buffer
if sqlErr != nil {
if that.testTx != nil {
if that.txFailed != nil {
*that.txFailed = true
}
return resl, that.LastErr
return resl, sqlErr
}
// 事务内不做连接级重试：死锁等错误会导致 MySQL 自动回滚事务，
// 在已回滚的 Tx 上重试会以 auto-commit 模式执行，造成数据不一致
if that.Tx != nil {
if that.txFailed != nil {
*that.txFailed = true
}
return resl, that.LastErr
return resl, sqlErr
}
if !retried {
if pingErr := that.DB.Ping(); pingErr == nil {
sqlErr = nil
return that.execWithRetry(query, true, args...)
}
}
return resl, that.LastErr
return resl, sqlErr
}
return resl, that.LastErr
return resl, sqlErr
}
// logSQL 统一 SQL 日志输出（仅此一处打印，消除重复）
func (that *HoTimeDB) logSQL(query string, args []interface{}, err error) {
if err != nil {
msg := fmt.Sprintf("SQL: %s DATA: %v ERROR: %v", query, args, err)
that.testLogBufMu.Lock()
if that.testLogBuf != nil {
that.testLogBuf.WriteString(msg + "\n")
} else if that.Log != nil {
that.Log.Error().Msg(msg)
}
that.testLogBufMu.Unlock()
return
}
// 无错误时只在 logLevel > 0 时打印 DEBUG 级别
if that.Log != nil && that.Log.GetLevel() > 0 {
msg := fmt.Sprintf("SQL: %s DATA: %v", query, args)
that.testLogBufMu.Lock()
if that.testLogBuf != nil {
that.testLogBuf.WriteString(msg + "\n")
} else {
that.Log.Debug().Msg(msg)
}
that.testLogBufMu.Unlock()
}
}
// processArgs 处理参数中的 slice 类型
@@ -587,7 +575,9 @@ func (that *HoTimeDB) Row(resl *sql.Rows) []Map {
b[j] = &a[j]
}
if err := resl.Scan(b...); err != nil {
that.LastErr.SetError(err)
dbErr := &DBError{}
dbErr.SetError(err)
that.lastError.Store(dbErr)
return nil
}
for j := 0; j < colCount; j++ {
@@ -615,7 +605,9 @@ func (that *HoTimeDB) Row(resl *sql.Rows) []Map {
dest = append(dest, lis)
}
if err := resl.Err(); err != nil {
that.LastErr.SetError(err)
dbErr := &DBError{}
dbErr.SetError(err)
that.lastError.Store(dbErr)
}
return dest
}
db/transaction.go
+9 -9
View File
@@ -23,16 +23,11 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce
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,
Dialect: that.Dialect,
mu: sync.RWMutex{},
limitMu: sync.Mutex{},
testTx: that.testTx,
testMu: that.testMu,
@@ -76,7 +71,9 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce
tx, err := db.Begin()
if err != nil {
that.LastErr.SetError(err)
dbErr := &DBError{}
dbErr.SetError(err)
that.lastError.Store(dbErr)
return isSuccess
}
@@ -84,7 +81,6 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce
isSuccess = action(db)
// SQL 出过错 → 事务必须回滚，不管回调返回什么
if txFailed {
_ = db.Tx.Rollback()
return false
@@ -93,7 +89,9 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce
if !isSuccess {
err = db.Tx.Rollback()
if err != nil {
that.LastErr.SetError(err)
dbErr := &DBError{}
dbErr.SetError(err)
that.lastError.Store(dbErr)
return isSuccess
}
return isSuccess
@@ -101,7 +99,9 @@ func (that *HoTimeDB) Action(action func(db HoTimeDB) (isSuccess bool)) (isSucce
err = db.Tx.Commit()
if err != nil {
that.LastErr.SetError(err)
dbErr := &DBError{}
dbErr.SetError(err)
that.lastError.Store(dbErr)
return false
}
return true
db/transaction_test.go
+2 -6
View File
@@ -2,11 +2,10 @@ package db
import (
. "code.hoteas.com/golang/hotime/common"
"code.hoteas.com/golang/hotime/log"
"database/sql"
"sync"
"testing"
"github.com/sirupsen/logrus"
)
func newTestDB(t *testing.T) *HoTimeDB {
@@ -15,14 +14,11 @@ func newTestDB(t *testing.T) *HoTimeDB {
if err != nil {
t.Fatal(err)
}
logger := logrus.New()
logger.SetLevel(logrus.WarnLevel)
db := &HoTimeDB{
DB: sqlDB,
Type: "sqlite3",
Dialect: &SQLiteDialect{},
LastErr: &Error{Logger: logger},
Log: logger,
Log: log.NewTestLogger(),
}
_, execErr := sqlDB.Exec("CREATE TABLE test_item (id INTEGER PRIMARY KEY, name TEXT, value INTEGER)")
if execErr != nil {