600 lines
13 KiB
Go
600 lines
13 KiB
Go
// 数据库工具包
|
||
package db
|
||
|
||
import (
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"reflect"
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
"background/logs"
|
||
)
|
||
|
||
// 数据容器抽象对象定义
|
||
type Database struct {
|
||
Type string // 用来给SqlBuilder进行一些特殊的判断 (空值或mysql 皆表示这是一个MySQL实例)
|
||
DB *sql.DB
|
||
}
|
||
|
||
// SQL异步执行队列定义
|
||
type queueList struct {
|
||
list []*QueueItem //队列列表
|
||
sleeping chan bool
|
||
loop chan bool
|
||
lock sync.RWMutex
|
||
quit chan bool
|
||
quited bool
|
||
}
|
||
|
||
// SQL异步执行队列子元素定义
|
||
type QueueItem struct {
|
||
DB *Database //数据库对象
|
||
Query string //SQL语句字符串
|
||
Params []interface{} //参数列表
|
||
}
|
||
|
||
// 缓存数据对象定义
|
||
type cache struct {
|
||
data map[string]map[string]interface{}
|
||
}
|
||
|
||
func (this *cache) Init() {
|
||
this.data["default"] = make(map[string]interface{})
|
||
}
|
||
|
||
// 设置缓存
|
||
func (this *cache) Set(key string, value interface{}, args ...string) {
|
||
var group string
|
||
if len(args) > 0 {
|
||
group = args[0]
|
||
if _, exist := this.data[group]; !exist {
|
||
this.data[group] = make(map[string]interface{})
|
||
}
|
||
} else {
|
||
group = "default"
|
||
}
|
||
this.data[group][key] = value
|
||
}
|
||
|
||
// 获取缓存数据
|
||
func (this *cache) Get(key string, args ...string) interface{} {
|
||
var group string
|
||
if len(args) > 0 {
|
||
group = args[0]
|
||
} else {
|
||
group = "default"
|
||
}
|
||
if g, exist := this.data[group]; exist {
|
||
if v, ok := g[key]; ok {
|
||
return v
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 删除缓存数据
|
||
func (this *cache) Del(key string, args ...string) {
|
||
var group string
|
||
if len(args) > 0 {
|
||
group = args[0]
|
||
} else {
|
||
group = "default"
|
||
}
|
||
if g, exist := this.data[group]; exist {
|
||
if _, ok := g[key]; ok {
|
||
delete(this.data[group], key)
|
||
}
|
||
}
|
||
}
|
||
|
||
var (
|
||
lastError error
|
||
Cache *cache
|
||
queue *queueList
|
||
Obj *Database
|
||
)
|
||
|
||
func init() {
|
||
Cache = &cache{data: make(map[string]map[string]interface{})}
|
||
Cache.Init()
|
||
queue = &queueList{}
|
||
go queue.Start()
|
||
}
|
||
|
||
// 关闭数据库连接
|
||
func (this *Database) Close() {
|
||
this.DB.Close()
|
||
}
|
||
|
||
// 获取最后发生的错误字符串
|
||
func LastErr() string {
|
||
if lastError != nil {
|
||
return lastError.Error()
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// 执行语句
|
||
func (this *Database) Exec(query string, args ...interface{}) (sql.Result, error) {
|
||
return this.DB.Exec(query, args...)
|
||
}
|
||
|
||
// 查询单条记录
|
||
func (this *Database) Query(query string, args ...interface{}) (*sql.Rows, error) {
|
||
return this.DB.Query(query, args...)
|
||
}
|
||
|
||
// 查询单条记录
|
||
func (this *Database) QueryRow(query string, args ...interface{}) *sql.Row {
|
||
return this.DB.QueryRow(query, args...)
|
||
}
|
||
|
||
// Query2 查询实体集合
|
||
// obj 为接收数据的实体指针
|
||
func (this *Database) Query2(sql string, obj interface{}, args ...interface{}) error {
|
||
var tagMap map[string]int
|
||
var tp, tps reflect.Type
|
||
var n, i int
|
||
var err error
|
||
var ret reflect.Value
|
||
// 检测val参数是否为我们所想要的参数
|
||
tp = reflect.TypeOf(obj)
|
||
if reflect.Ptr != tp.Kind() {
|
||
return errors.New("is not pointer")
|
||
}
|
||
|
||
if reflect.Slice != tp.Elem().Kind() {
|
||
return errors.New("is not slice pointer")
|
||
}
|
||
|
||
tp = tp.Elem()
|
||
tps = tp.Elem()
|
||
if reflect.Struct != tps.Kind() {
|
||
return errors.New("is not struct slice pointer")
|
||
}
|
||
|
||
tagMap = make(map[string]int)
|
||
n = tps.NumField()
|
||
for i = 0; i < n; i++ {
|
||
tag := tps.Field(i).Tag.Get("sql")
|
||
if len(tag) > 0 {
|
||
tagMap[tag] = i + 1
|
||
}
|
||
}
|
||
|
||
// 执行查询
|
||
ret, err = this.queryAndReflect(sql, tagMap, tp, args...)
|
||
if nil != err {
|
||
return err
|
||
}
|
||
|
||
// 返回结果
|
||
reflect.ValueOf(obj).Elem().Set(ret)
|
||
|
||
return nil
|
||
}
|
||
|
||
// queryAndReflect 查询并将结果反射成实体集合
|
||
func (this *Database) queryAndReflect(sql string,
|
||
tagMap map[string]int,
|
||
tpSlice reflect.Type, args ...interface{}) (reflect.Value, error) {
|
||
|
||
var ret reflect.Value
|
||
|
||
// 执行sql语句
|
||
rows, err := this.DB.Query(sql, args...)
|
||
if nil != err {
|
||
return reflect.Value{}, err
|
||
}
|
||
|
||
defer rows.Close()
|
||
// 开始枚举结果
|
||
cols, err := rows.Columns()
|
||
if nil != err {
|
||
return reflect.Value{}, err
|
||
}
|
||
|
||
ret = reflect.MakeSlice(tpSlice, 0, 50)
|
||
// 构建接收队列
|
||
scan := make([]interface{}, len(cols))
|
||
row := make([]interface{}, len(cols))
|
||
for r := range row {
|
||
scan[r] = &row[r]
|
||
}
|
||
|
||
for rows.Next() {
|
||
feild := reflect.New(tpSlice.Elem()).Elem()
|
||
// 取得结果
|
||
err = rows.Scan(scan...)
|
||
// 开始遍历结果
|
||
for i := 0; i < len(cols); i++ {
|
||
n := tagMap[cols[i]] - 1
|
||
|
||
if n < 0 {
|
||
continue
|
||
}
|
||
switch feild.Type().Field(n).Type.Kind() {
|
||
case reflect.Bool:
|
||
if nil != row[i] {
|
||
feild.Field(n).SetBool("false" != string(row[i].([]byte)))
|
||
} else {
|
||
feild.Field(n).SetBool(false)
|
||
}
|
||
case reflect.String:
|
||
if nil != row[i] {
|
||
feild.Field(n).SetString(string(row[i].([]byte)))
|
||
} else {
|
||
feild.Field(n).SetString("")
|
||
}
|
||
case reflect.Float32:
|
||
if nil != row[i] {
|
||
//log.Println(row[i].(float32))
|
||
switch reflect.TypeOf(row[i]).Kind() {
|
||
case reflect.Slice:
|
||
v, e := strconv.ParseFloat(string(row[i].([]byte)), 0)
|
||
if nil == e {
|
||
feild.Field(n).SetFloat(float64(v))
|
||
//feild.Field(n).SetFloat(float64(row[i].(float32)))
|
||
}
|
||
break
|
||
case reflect.Float64:
|
||
feild.Field(n).SetFloat(float64(row[i].(float32)))
|
||
}
|
||
|
||
} else {
|
||
feild.Field(n).SetFloat(0)
|
||
}
|
||
case reflect.Float64:
|
||
if nil != row[i] {
|
||
//log.Println(row[i].(float32))
|
||
//v, e := strconv.ParseFloat(string(row[i].([]byte)), 0)
|
||
//if nil == e {
|
||
feild.Field(n).SetFloat(row[i].(float64))
|
||
//}
|
||
} else {
|
||
feild.Field(n).SetFloat(0)
|
||
}
|
||
case reflect.Int8:
|
||
fallthrough
|
||
case reflect.Int16:
|
||
fallthrough
|
||
case reflect.Int32:
|
||
fallthrough
|
||
case reflect.Int64:
|
||
fallthrough
|
||
case reflect.Int:
|
||
if nil != row[i] {
|
||
byRow, ok := row[i].([]byte)
|
||
if ok {
|
||
v, e := strconv.ParseInt(string(byRow), 10, 64)
|
||
if nil == e {
|
||
feild.Field(n).SetInt(v)
|
||
}
|
||
} else {
|
||
v, e := strconv.ParseInt(fmt.Sprint(row[i]), 10, 64)
|
||
if nil == e {
|
||
feild.Field(n).SetInt(v)
|
||
}
|
||
}
|
||
} else {
|
||
feild.Field(n).SetInt(0)
|
||
}
|
||
}
|
||
}
|
||
|
||
ret = reflect.Append(ret, feild)
|
||
}
|
||
|
||
return ret, nil
|
||
}
|
||
|
||
// 执行UPDATE语句并返回受影响的行数
|
||
// 返回0表示没有出错, 但没有被更新的行
|
||
// 返回-1表示出错
|
||
func (this *Database) Update(query string, args ...interface{}) (int64, error) {
|
||
ret, err := this.Exec(query, args...)
|
||
if err != nil {
|
||
return -1, err
|
||
}
|
||
aff, err := ret.RowsAffected()
|
||
if err != nil {
|
||
return -1, err
|
||
}
|
||
return aff, nil
|
||
}
|
||
|
||
// 执行DELETE语句并返回受影响的行数
|
||
// 返回0表示没有出错, 但没有被删除的行
|
||
// 返回-1表示出错
|
||
func (this *Database) Delete(query string, args ...interface{}) (int64, error) {
|
||
return this.Update(query, args...)
|
||
}
|
||
|
||
func GenSql(obj interface{}) (string, error) {
|
||
ret := ""
|
||
typ := reflect.TypeOf(obj).Kind()
|
||
if typ != reflect.Struct {
|
||
return (""), errors.New("not a struct")
|
||
}
|
||
value := obj.(reflect.Value)
|
||
num := value.NumField()
|
||
for i := 0; i < num; i++ {
|
||
if i == 0 {
|
||
ret += "("
|
||
}
|
||
switch value.Field(i).Type().Kind() {
|
||
case reflect.String:
|
||
str := value.Field(i).Interface().(string)
|
||
if str[0] != '"' {
|
||
ret += "\""
|
||
str += "\""
|
||
ret += str
|
||
|
||
} else {
|
||
ret += value.Field(i).Interface().(string)
|
||
}
|
||
case reflect.Int:
|
||
ret += fmt.Sprintf("%d", value.Field(i).Interface().(int))
|
||
case reflect.Int8:
|
||
ret += fmt.Sprintf("%d", value.Field(i).Interface().(int8))
|
||
case reflect.Int32:
|
||
ret += fmt.Sprintf("%d", value.Field(i).Interface().(int32))
|
||
case reflect.Int64:
|
||
ret += fmt.Sprintf("%d", value.Field(i).Interface().(int64))
|
||
case reflect.Int16:
|
||
ret += fmt.Sprintf("%d", value.Field(i).Interface().(int16))
|
||
case reflect.Bool:
|
||
if value.Field(i).Interface().(bool) {
|
||
ret += fmt.Sprintf("true")
|
||
} else {
|
||
ret += fmt.Sprintf("false")
|
||
}
|
||
case reflect.Float32:
|
||
ret += fmt.Sprintf("%x", value.Field(i).Interface().(float32))
|
||
case reflect.Float64:
|
||
ret += fmt.Sprintf("true", value.Field(i).Interface().(float64))
|
||
}
|
||
if i == num-1 {
|
||
ret += ")"
|
||
} else {
|
||
ret += ","
|
||
}
|
||
}
|
||
return ret, nil
|
||
}
|
||
func (this *Database) InsertObejct(tb_name string, obj interface{}) (int64, error) {
|
||
var tagMap map[int]string
|
||
var tp, tps reflect.Type
|
||
var n, i int
|
||
|
||
// 检测val参数是否为我们所想要的参数
|
||
tp = reflect.TypeOf(obj)
|
||
if reflect.Ptr != tp.Kind() {
|
||
return 0, errors.New("is not pointer")
|
||
}
|
||
|
||
if reflect.Slice != tp.Elem().Kind() {
|
||
return 0, errors.New("is not slice pointer")
|
||
}
|
||
|
||
tp = tp.Elem()
|
||
tps = tp.Elem()
|
||
value := reflect.ValueOf(obj).Elem()
|
||
|
||
if reflect.Struct != tps.Kind() {
|
||
return 0, errors.New("is not struct slice pointer")
|
||
}
|
||
for z := 0; z < value.Len(); z++ {
|
||
tagMap = make(map[int]string)
|
||
n = tps.NumField()
|
||
var query_struct string
|
||
for i = 0; i < n; i++ {
|
||
tag := tps.Field(i).Tag.Get("sql")
|
||
if len(tag) > 0 {
|
||
tagMap[i] = tag
|
||
}
|
||
if i == 0 {
|
||
query_struct += "("
|
||
}
|
||
query_struct += tagMap[i]
|
||
if i == n-1 {
|
||
query_struct += ")"
|
||
} else {
|
||
query_struct += ","
|
||
}
|
||
}
|
||
vs, e := GenSql(value.Index(z))
|
||
if nil != e {
|
||
logs.Error(e.Error())
|
||
}
|
||
query := "insert into " + tb_name + query_struct + "values " + vs
|
||
_, e = this.Insert(query)
|
||
if e != nil {
|
||
logs.Error(e.Error())
|
||
}
|
||
}
|
||
return 0, nil
|
||
}
|
||
|
||
// 执行INSERT语句并返回最后生成的自增ID
|
||
// 返回0表示没有出错, 但没生成自增ID
|
||
// 返回-1表示出错
|
||
func (this *Database) Insert(query string, args ...interface{}) (int64, error) {
|
||
ret, err := this.Exec(query, args...)
|
||
if err != nil {
|
||
return -1, err
|
||
}
|
||
last, err := ret.LastInsertId()
|
||
if err != nil {
|
||
return -1, err
|
||
|
||
}
|
||
return last, nil
|
||
}
|
||
|
||
type OneRow map[string]string
|
||
type Results []OneRow
|
||
|
||
// 判断字段是否存在
|
||
func (row OneRow) Exist(field string) bool {
|
||
if _, ok := row[field]; ok {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
// 获取指定字段的值
|
||
func (row OneRow) Get(field string) string {
|
||
if v, ok := row[field]; ok {
|
||
return v
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// 获取指定字段的整数值, 注意, 如果该字段不存在则会返回0
|
||
func (row OneRow) GetInt(field string) int {
|
||
if v, ok := row[field]; ok {
|
||
return Atoi(v)
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// 获取指定字段的整数值, 注意, 如果该字段不存在则会返回0
|
||
func (row OneRow) GetInt64(field string) int64 {
|
||
if v, ok := row[field]; ok {
|
||
return Atoi64(v)
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// 设置值
|
||
func (row OneRow) Set(key, val string) {
|
||
row[key] = val
|
||
}
|
||
|
||
// 查询不定字段的结果集
|
||
func (this *Database) Select(query string, args ...interface{}) (Results, error) {
|
||
rows, err := this.DB.Query(query, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
cols, err := rows.Columns()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
colNum := len(cols)
|
||
rawValues := make([][]byte, colNum)
|
||
scans := make([]interface{}, len(cols)) //query.Scan的参数，因为每次查询出来的列是不定长的，所以传入长度固定当次查询的长度
|
||
|
||
// 将每行数据填充到[][]byte里
|
||
for i := range rawValues {
|
||
scans[i] = &rawValues[i]
|
||
}
|
||
|
||
results := make(Results, 0)
|
||
for rows.Next() {
|
||
err = rows.Scan(scans...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
row := make(map[string]string)
|
||
|
||
for k, raw := range rawValues {
|
||
key := cols[k]
|
||
/*if raw == nil {
|
||
row[key] = "\\N"
|
||
} else {*/
|
||
row[key] = string(raw)
|
||
//}
|
||
}
|
||
results = append(results, row)
|
||
}
|
||
return results, nil
|
||
}
|
||
|
||
// 查询一行不定字段的结果
|
||
func (this *Database) SelectOne(query string, args ...interface{}) (OneRow, error) {
|
||
ret, err := this.Select(query, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(ret) > 0 {
|
||
return ret[0], nil
|
||
}
|
||
return make(OneRow), nil
|
||
}
|
||
|
||
// 队列入栈
|
||
func (this *queueList) Push(item *QueueItem) {
|
||
this.lock.Lock()
|
||
this.list = append(this.list, item)
|
||
this.lock.Unlock()
|
||
}
|
||
|
||
// 队列出栈
|
||
func (this *queueList) Pop() chan *QueueItem {
|
||
item := make(chan *QueueItem)
|
||
go func() {
|
||
defer close(item)
|
||
for {
|
||
switch {
|
||
case len(this.list) == 0:
|
||
timeout := time.After(time.Second * 2)
|
||
select {
|
||
case <-this.quit:
|
||
this.quited = true
|
||
return
|
||
case <-timeout:
|
||
//log.Println("SQL Queue polling")
|
||
}
|
||
default:
|
||
this.lock.Lock()
|
||
i := this.list[0]
|
||
this.list = this.list[1:]
|
||
this.lock.Unlock()
|
||
select {
|
||
case item <- i:
|
||
return
|
||
case <-this.quit:
|
||
this.quited = true
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
return item
|
||
}
|
||
|
||
// 执行开始执行
|
||
func (this *queueList) Start() {
|
||
for {
|
||
if this.quited {
|
||
return
|
||
}
|
||
c := this.Pop()
|
||
item := <-c
|
||
item.DB.Exec(item.Query, item.Params...)
|
||
}
|
||
}
|
||
|
||
// 停止队列
|
||
func (this *queueList) Stop() {
|
||
this.quit <- true
|
||
}
|
||
|
||
// 向Sql队列中插入一条执行语句
|
||
func (this *Database) Queue(query string, args ...interface{}) {
|
||
item := &QueueItem{
|
||
DB: this,
|
||
Query: query,
|
||
Params: args,
|
||
}
|
||
queue.Push(item)
|
||
}
|