hoteas
87098a9180
- 在 HoTimeDB 中新增 testMu 互斥锁，确保 testTx 的串行访问，解决并发冲突问题 - 在 BeginTestTx 中初始化 testMu，确保测试事务的安全性 - 在 HoTimeCache 中新增 DisableDbCache 方法，测试模式下禁用 DB 和 Redis 缓存，避免锁等待超时 - 更新 Select 方法，确保在 testTx 激活时跳过缓存逻辑，提升测试稳定性 - 优化 Swagger 生成逻辑，支持模块化输出和导航页生成 - 移除冗余的调试日志代码，提升代码整洁性
572 lines
15 KiB
Go
572 lines
15 KiB
Go
package db
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"math"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
|
|
. "code.hoteas.com/golang/hotime/common"
|
|
)
|
|
|
|
// md5 生成查询的 MD5 哈希（用于缓存）
|
|
func (that *HoTimeDB) md5(query string, args ...interface{}) string {
|
|
strByte, _ := json.Marshal(args)
|
|
str := Md5(query + ":" + string(strByte))
|
|
return str
|
|
}
|
|
|
|
// Query 执行查询 SQL
|
|
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 {
|
|
// 预处理数组占位符 ?[]
|
|
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()
|
|
that.Log.Info("SQL:"+that.LastQuery, " DATA:", that.LastData, " ERROR:", that.LastErr.GetError())
|
|
that.mu.RUnlock()
|
|
}
|
|
}()
|
|
|
|
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)
|
|
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 {
|
|
if that.testMu != nil { that.testMu.Unlock() }
|
|
return nil
|
|
}
|
|
result := that.Row(resl)
|
|
if that.testMu != nil { that.testMu.Unlock() }
|
|
return result
|
|
} else if that.Tx != nil {
|
|
resl, err = that.Tx.Query(query, processedArgs...)
|
|
} else {
|
|
resl, err = db.Query(query, processedArgs...)
|
|
}
|
|
|
|
that.LastErr.SetError(err)
|
|
if err != nil {
|
|
if !retried {
|
|
if pingErr := db.Ping(); pingErr == nil {
|
|
return that.queryWithRetry(query, true, args...)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
return that.Row(resl)
|
|
}
|
|
|
|
// Exec 执行非查询 SQL
|
|
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) {
|
|
// 预处理数组占位符 ?[]
|
|
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()
|
|
that.Log.Info("SQL: "+that.LastQuery, " DATA: ", that.LastData, " ERROR: ", that.LastErr.GetError())
|
|
that.mu.RUnlock()
|
|
}
|
|
}()
|
|
|
|
var e error
|
|
var resl sql.Result
|
|
|
|
if that.DB == nil {
|
|
err := errors.New("没有初始化数据库")
|
|
that.LastErr.SetError(err)
|
|
return nil, that.LastErr
|
|
}
|
|
|
|
// 处理参数中的 slice 类型
|
|
processedArgs := that.processArgs(args)
|
|
|
|
if that.testTx != nil {
|
|
if that.testMu != nil { that.testMu.Lock() }
|
|
resl, e = that.testTx.Exec(query, processedArgs...)
|
|
if that.testMu != nil { that.testMu.Unlock() }
|
|
} else if that.Tx != nil {
|
|
resl, e = that.Tx.Exec(query, processedArgs...)
|
|
} else {
|
|
resl, e = that.DB.Exec(query, processedArgs...)
|
|
}
|
|
|
|
that.LastErr.SetError(e)
|
|
if e != nil {
|
|
// testTx 模式下不走连接池重试，避免 busy buffer
|
|
if that.testTx != nil {
|
|
return resl, that.LastErr
|
|
}
|
|
if !retried {
|
|
if pingErr := that.DB.Ping(); pingErr == nil {
|
|
return that.execWithRetry(query, true, args...)
|
|
}
|
|
}
|
|
return resl, that.LastErr
|
|
}
|
|
|
|
return resl, that.LastErr
|
|
}
|
|
|
|
// processArgs 处理参数中的 slice 类型
|
|
func (that *HoTimeDB) processArgs(args []interface{}) []interface{} {
|
|
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)
|
|
// 将slice转为逗号分割字符串
|
|
argStr := ""
|
|
for i := 0; i < len(argLis); i++ {
|
|
if i == len(argLis)-1 {
|
|
argStr += ObjToStr(argLis[i])
|
|
} else {
|
|
argStr += ObjToStr(argLis[i]) + ","
|
|
}
|
|
}
|
|
processedArgs[key] = argStr
|
|
}
|
|
}
|
|
return processedArgs
|
|
}
|
|
|
|
// expandArrayPlaceholder 展开 IN (?) / NOT IN (?) 中的数组参数
|
|
// 自动识别 IN/NOT IN (?) 模式，当参数是数组时展开为多个 ?
|
|
//
|
|
// 示例:
|
|
//
|
|
// db.Query("SELECT * FROM user WHERE id IN (?)", []int{1, 2, 3})
|
|
// // 展开为: SELECT * FROM user WHERE id IN (?, ?, ?) 参数: [1, 2, 3]
|
|
//
|
|
// db.Query("SELECT * FROM user WHERE id IN (?)", []int{})
|
|
// // 展开为: SELECT * FROM user WHERE 1=0 参数: [] (空集合的IN永假)
|
|
//
|
|
// db.Query("SELECT * FROM user WHERE id NOT IN (?)", []int{})
|
|
// // 展开为: SELECT * FROM user WHERE 1=1 参数: [] (空集合的NOT IN永真)
|
|
//
|
|
// db.Query("SELECT * FROM user WHERE id = ?", 1)
|
|
// // 保持不变: SELECT * FROM user WHERE id = ? 参数: [1]
|
|
func (that *HoTimeDB) expandArrayPlaceholder(query string, args []interface{}) (string, []interface{}) {
|
|
if len(args) == 0 || !strings.Contains(query, "?") {
|
|
return query, args
|
|
}
|
|
|
|
// 检查是否有数组参数
|
|
hasArray := false
|
|
for _, arg := range args {
|
|
if arg == nil {
|
|
continue
|
|
}
|
|
argType := reflect.ValueOf(arg).Type().String()
|
|
if strings.Contains(argType, "[]") || strings.Contains(argType, "Slice") {
|
|
hasArray = true
|
|
break
|
|
}
|
|
}
|
|
if !hasArray {
|
|
return query, args
|
|
}
|
|
|
|
newArgs := make([]interface{}, 0, len(args))
|
|
result := strings.Builder{}
|
|
argIndex := 0
|
|
|
|
for i := 0; i < len(query); i++ {
|
|
if query[i] == '?' && argIndex < len(args) {
|
|
arg := args[argIndex]
|
|
argIndex++
|
|
|
|
if arg == nil {
|
|
result.WriteByte('?')
|
|
newArgs = append(newArgs, nil)
|
|
continue
|
|
}
|
|
|
|
argType := reflect.ValueOf(arg).Type().String()
|
|
if strings.Contains(argType, "[]") || strings.Contains(argType, "Slice") {
|
|
// 是数组参数，检查是否在 IN (...) 或 NOT IN (...) 中
|
|
prevPart := result.String()
|
|
prevUpper := strings.ToUpper(prevPart)
|
|
|
|
// 查找最近的 NOT IN ( 模式
|
|
notInIndex := strings.LastIndex(prevUpper, " NOT IN (")
|
|
notInIndex2 := strings.LastIndex(prevUpper, " NOT IN(")
|
|
if notInIndex2 > notInIndex {
|
|
notInIndex = notInIndex2
|
|
}
|
|
|
|
// 查找最近的 IN ( 模式（但要排除 NOT IN 的情况）
|
|
inIndex := strings.LastIndex(prevUpper, " IN (")
|
|
inIndex2 := strings.LastIndex(prevUpper, " IN(")
|
|
if inIndex2 > inIndex {
|
|
inIndex = inIndex2
|
|
}
|
|
|
|
// 判断是 NOT IN 还是 IN
|
|
// 注意：" NOT IN (" 包含 " IN ("，所以如果找到的 IN 位置在 NOT IN 范围内，应该优先判断为 NOT IN
|
|
isNotIn := false
|
|
matchIndex := -1
|
|
if notInIndex != -1 {
|
|
// 检查 inIndex 是否在 notInIndex 范围内（即 NOT IN 的 IN 部分）
|
|
// NOT IN ( 的 IN ( 部分从 notInIndex + 4 开始
|
|
if inIndex != -1 && inIndex >= notInIndex && inIndex <= notInIndex+5 {
|
|
// inIndex 是 NOT IN 的一部分，使用 NOT IN
|
|
isNotIn = true
|
|
matchIndex = notInIndex
|
|
} else if inIndex == -1 || notInIndex > inIndex {
|
|
// 没有独立的 IN，或 NOT IN 在 IN 之后
|
|
isNotIn = true
|
|
matchIndex = notInIndex
|
|
} else {
|
|
// 有独立的 IN 且在 NOT IN 之后
|
|
matchIndex = inIndex
|
|
}
|
|
} else if inIndex != -1 {
|
|
matchIndex = inIndex
|
|
}
|
|
|
|
// 检查 IN ( 后面是否只有空格（即当前 ? 紧跟在 IN ( 后面）
|
|
isInPattern := false
|
|
if matchIndex != -1 {
|
|
afterIn := prevPart[matchIndex:]
|
|
// 找到 ( 的位置
|
|
parenIdx := strings.Index(afterIn, "(")
|
|
if parenIdx != -1 {
|
|
afterParen := strings.TrimSpace(afterIn[parenIdx+1:])
|
|
if afterParen == "" {
|
|
isInPattern = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if isInPattern {
|
|
// 在 IN (...) 或 NOT IN (...) 模式中
|
|
argList := ObjToSlice(arg)
|
|
if len(argList) == 0 {
|
|
// 空数组处理：需要找到字段名的开始位置
|
|
// 往前找最近的 AND/OR/WHERE/(，以确定条件的开始位置
|
|
truncateIndex := matchIndex
|
|
searchPart := prevUpper[:matchIndex]
|
|
|
|
// 找最近的分隔符位置
|
|
andIdx := strings.LastIndex(searchPart, " AND ")
|
|
orIdx := strings.LastIndex(searchPart, " OR ")
|
|
whereIdx := strings.LastIndex(searchPart, " WHERE ")
|
|
parenIdx := strings.LastIndex(searchPart, "(")
|
|
|
|
// 取最靠后的分隔符
|
|
sepIndex := -1
|
|
sepLen := 0
|
|
if andIdx > sepIndex {
|
|
sepIndex = andIdx
|
|
sepLen = 5 // " AND "
|
|
}
|
|
if orIdx > sepIndex {
|
|
sepIndex = orIdx
|
|
sepLen = 4 // " OR "
|
|
}
|
|
if whereIdx > sepIndex {
|
|
sepIndex = whereIdx
|
|
sepLen = 7 // " WHERE "
|
|
}
|
|
if parenIdx > sepIndex {
|
|
sepIndex = parenIdx
|
|
sepLen = 1 // "("
|
|
}
|
|
|
|
if sepIndex != -1 {
|
|
truncateIndex = sepIndex + sepLen
|
|
}
|
|
|
|
result.Reset()
|
|
result.WriteString(prevPart[:truncateIndex])
|
|
if isNotIn {
|
|
// NOT IN 空集合 = 永真
|
|
result.WriteString(" 1=1 ")
|
|
} else {
|
|
// IN 空集合 = 永假
|
|
result.WriteString(" 1=0 ")
|
|
}
|
|
// 跳过后面的 )
|
|
for j := i + 1; j < len(query); j++ {
|
|
if query[j] == ')' {
|
|
i = j
|
|
break
|
|
}
|
|
}
|
|
} else if len(argList) == 1 {
|
|
// 单元素数组
|
|
result.WriteByte('?')
|
|
newArgs = append(newArgs, argList[0])
|
|
} else {
|
|
// 多元素数组，展开为多个 ?
|
|
for j := 0; j < len(argList); j++ {
|
|
if j > 0 {
|
|
result.WriteString(", ")
|
|
}
|
|
result.WriteByte('?')
|
|
newArgs = append(newArgs, argList[j])
|
|
}
|
|
}
|
|
} else {
|
|
// 不在 IN 模式中，保持原有行为（数组会被 processArgs 转为逗号字符串）
|
|
result.WriteByte('?')
|
|
newArgs = append(newArgs, arg)
|
|
}
|
|
} else {
|
|
// 非数组参数
|
|
result.WriteByte('?')
|
|
newArgs = append(newArgs, arg)
|
|
}
|
|
} else {
|
|
result.WriteByte(query[i])
|
|
}
|
|
}
|
|
|
|
return result.String(), newArgs
|
|
}
|
|
|
|
const (
|
|
dbTypeUnknown = iota
|
|
dbTypeInteger
|
|
dbTypeFloat
|
|
dbTypeDecimal
|
|
)
|
|
|
|
// classifyDBType 将数据库类型名分类，覆盖 MySQL/SQLite/PostgreSQL
|
|
func classifyDBType(typeName string) int {
|
|
t := strings.ToUpper(strings.TrimSpace(typeName))
|
|
t = strings.TrimSuffix(strings.TrimSuffix(t, " UNSIGNED"), " SIGNED")
|
|
t = strings.TrimSpace(t)
|
|
base := t
|
|
if idx := strings.IndexByte(t, '('); idx > 0 {
|
|
base = t[:idx]
|
|
}
|
|
switch base {
|
|
case "INT", "INTEGER", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT",
|
|
"INT2", "INT4", "INT8",
|
|
"SERIAL", "BIGSERIAL", "SMALLSERIAL",
|
|
"YEAR", "BOOL", "BOOLEAN", "OID":
|
|
return dbTypeInteger
|
|
case "DECIMAL", "NUMERIC", "NEWDECIMAL":
|
|
return dbTypeDecimal
|
|
case "FLOAT", "DOUBLE", "REAL", "FLOAT4", "FLOAT8", "DOUBLE PRECISION":
|
|
return dbTypeFloat
|
|
}
|
|
if strings.Contains(base, "INT") {
|
|
return dbTypeInteger
|
|
}
|
|
if strings.Contains(base, "DECIMAL") || strings.Contains(base, "NUMERIC") {
|
|
return dbTypeDecimal
|
|
}
|
|
if strings.Contains(base, "DOUBLE") || strings.Contains(base, "FLOAT") {
|
|
return dbTypeFloat
|
|
}
|
|
return dbTypeUnknown
|
|
}
|
|
|
|
// getDecimalScale 提取 DECIMAL/NUMERIC 精度，失败返回 -1
|
|
func getDecimalScale(colType *sql.ColumnType, typeName string) int {
|
|
_, scale, ok := colType.DecimalSize()
|
|
if ok && scale >= 0 {
|
|
return int(scale)
|
|
}
|
|
// SQLite 不支持 DecimalSize，从类型名解析 "DECIMAL(10,2)" → 2
|
|
upper := strings.ToUpper(typeName)
|
|
if idx := strings.Index(upper, ","); idx > 0 {
|
|
rest := upper[idx+1:]
|
|
if end := strings.IndexByte(rest, ')'); end > 0 {
|
|
if s, err := strconv.Atoi(strings.TrimSpace(rest[:end])); err == nil {
|
|
return s
|
|
}
|
|
}
|
|
}
|
|
return -1
|
|
}
|
|
|
|
// getFloatScale 提取 FLOAT/DOUBLE 显式精度，无显式精度返回 -1 表示不截断
|
|
func getFloatScale(colType *sql.ColumnType, typeName string) int {
|
|
_, scale, ok := colType.DecimalSize()
|
|
if ok && scale > 0 && scale < 20 {
|
|
return int(scale)
|
|
}
|
|
return -1
|
|
}
|
|
|
|
// fixFloatValue 对 float64 值做类型感知精度修正
|
|
func fixFloatValue(f float64, category int, scale int) interface{} {
|
|
if math.IsNaN(f) || math.IsInf(f, 0) {
|
|
return float64(0)
|
|
}
|
|
switch category {
|
|
case dbTypeInteger:
|
|
return int64(math.Round(f))
|
|
case dbTypeDecimal:
|
|
return roundFloat(f, scale)
|
|
case dbTypeFloat:
|
|
return roundFloat(f, scale)
|
|
default:
|
|
return f
|
|
}
|
|
}
|
|
|
|
// roundFloat 使用 strconv 精确舍入，避免 f*pow 中间值产生精度误差
|
|
func roundFloat(f float64, scale int) interface{} {
|
|
if math.IsNaN(f) || math.IsInf(f, 0) {
|
|
return float64(0)
|
|
}
|
|
if scale == 0 {
|
|
// DECIMAL(10,0) 明确是整数
|
|
return math.Round(f)
|
|
}
|
|
if scale < 0 {
|
|
// 精度未知，不修正，原样返回
|
|
return f
|
|
}
|
|
s := strconv.FormatFloat(f, 'f', scale, 64)
|
|
v, err := strconv.ParseFloat(s, 64)
|
|
if err != nil {
|
|
return f
|
|
}
|
|
return v
|
|
}
|
|
|
|
// convertBytes 对 []byte 类型的列做类型感知解析
|
|
// MySQL 文本协议和 PG 的 DECIMAL/NUMERIC 以 []byte 返回，需要按列类型解析为正确 Go 类型
|
|
func convertBytes(raw []byte, category int, scale int) interface{} {
|
|
s := string(raw)
|
|
switch category {
|
|
case dbTypeInteger:
|
|
if n, err := strconv.ParseInt(s, 10, 64); err == nil {
|
|
return n
|
|
}
|
|
if f, err := strconv.ParseFloat(s, 64); err == nil {
|
|
return int64(math.Round(f))
|
|
}
|
|
return s
|
|
case dbTypeDecimal:
|
|
if f, err := strconv.ParseFloat(s, 64); err == nil {
|
|
return roundFloat(f, scale)
|
|
}
|
|
return s
|
|
case dbTypeFloat:
|
|
if f, err := strconv.ParseFloat(s, 64); err == nil {
|
|
return roundFloat(f, scale)
|
|
}
|
|
return s
|
|
default:
|
|
return s
|
|
}
|
|
}
|
|
|
|
// Row 数据库数据解析
|
|
func (that *HoTimeDB) Row(resl *sql.Rows) []Map {
|
|
defer resl.Close()
|
|
|
|
dest := make([]Map, 0)
|
|
strs, _ := resl.Columns()
|
|
colTypes, _ := resl.ColumnTypes()
|
|
|
|
colCount := len(strs)
|
|
categories := make([]int, colCount)
|
|
scales := make([]int, colCount)
|
|
if colTypes != nil {
|
|
for j := 0; j < len(colTypes) && j < colCount; j++ {
|
|
typeName := colTypes[j].DatabaseTypeName()
|
|
categories[j] = classifyDBType(typeName)
|
|
switch categories[j] {
|
|
case dbTypeDecimal:
|
|
scales[j] = getDecimalScale(colTypes[j], typeName)
|
|
case dbTypeFloat:
|
|
scales[j] = getFloatScale(colTypes[j], typeName)
|
|
}
|
|
}
|
|
}
|
|
|
|
for resl.Next() {
|
|
lis := make(Map, colCount)
|
|
a := make([]interface{}, colCount)
|
|
b := make([]interface{}, colCount)
|
|
for j := 0; j < colCount; j++ {
|
|
b[j] = &a[j]
|
|
}
|
|
if err := resl.Scan(b...); err != nil {
|
|
that.LastErr.SetError(err)
|
|
return nil
|
|
}
|
|
for j := 0; j < colCount; j++ {
|
|
val := a[j]
|
|
if val == nil {
|
|
lis[strs[j]] = nil
|
|
continue
|
|
}
|
|
switch v := val.(type) {
|
|
case []byte:
|
|
lis[strs[j]] = convertBytes(v, categories[j], scales[j])
|
|
case float64:
|
|
lis[strs[j]] = fixFloatValue(v, categories[j], scales[j])
|
|
case float32:
|
|
lis[strs[j]] = fixFloatValue(float64(v), categories[j], scales[j])
|
|
default:
|
|
lis[strs[j]] = val
|
|
}
|
|
}
|
|
dest = append(dest, lis)
|
|
}
|
|
if err := resl.Err(); err != nil {
|
|
that.LastErr.SetError(err)
|
|
}
|
|
return dest
|
|
}
|