Merge branch 'master' of https://github.com/duanhf2012/origin

cluster/cluster.go

@@ -5,6 +5,7 @@ import (
 	"math/rand"
 	"net"
 	"os"
+	"sort"
 	"strings"
 	"time"
 
@@ -274,7 +275,7 @@ func (slf *CCluster) GetNodeList(NodeServiceMethod string, rpcServerMethod *stri
 			servicename = servicename[1:]
 			nodeidList = append(nodeidList, GetNodeId())
 		} else {
-			nodeidList = slf.cfg.GetIdByService(servicename)
+			nodeidList = slf.cfg.GetIdByService(servicename, "")
 		}
 	} else {
 		nodeidList = slf.GetIdByNodeService(nodename, servicename)
@@ -293,7 +294,7 @@ func (slf *CCluster) GetNodeList(NodeServiceMethod string, rpcServerMethod *stri
 
 //GetNodeIdByServiceName 根据服务名查找nodeid servicename服务名 bOnline是否需要查找在线服务
 func (slf *CCluster) GetNodeIdByServiceName(servicename string, bOnline bool) []int {
-	nodeIDList := slf.cfg.GetIdByService(servicename)
+	nodeIDList := slf.cfg.GetIdByService(servicename, "")
 
 	if bOnline {
 		ret := make([]int, 0, len(nodeIDList))
@@ -308,6 +309,40 @@ func (slf *CCluster) GetNodeIdByServiceName(servicename string, bOnline bool) []
 	return nodeIDList
 }
 
+//根据Service获取负载均衡信息
+//负载均衡的策略是从配置获取所有配置了该服务的NodeId 并按NodeId排序 每个node负责处理数组index所在的那一部分
+func (slf *CCluster) GetBalancingInfo(currentNodeId int, servicename string, inSubNet bool) (*BalancingInfo, error) {
+	subNetName := ""
+	if inSubNet {
+		if node, ok := slf.cfg.mapIdNode[currentNodeId]; ok {
+			subNetName = node.SubNetName
+		} else {
+			return nil, fmt.Errorf("[cluster.GetBalancingInfo] cannot find node %d", currentNodeId)
+		}
+	}
+	lst := slf.cfg.GetIdByService(servicename, subNetName)
+	// if len(lst) <= 0 {
+	// 	return nil, fmt.Errorf("[cluster.GetBalancingInfo] cannot find service %s in any node", servicename)
+	// }
+	sort.Ints(lst)
+	ret := &BalancingInfo{
+		NodeId:      currentNodeId,
+		ServiceName: servicename,
+		TotalNum:    len(lst),
+		MyIndex:     -1,
+		NodeList:    lst,
+	}
+	if _, ok := slf.cfg.mapIdNode[currentNodeId]; ok {
+		for i, v := range lst {
+			if v == currentNodeId {
+				ret.MyIndex = i
+				break
+			}
+		}
+	}
+	return ret, nil
+}
+
 func (slf *CCluster) CheckNodeIsConnectedByID(nodeid int) bool {
 	if nodeid == GetNodeId() {
 		return true
@@ -522,6 +557,12 @@ func GetNodeIdByServiceName(serviceName string, bOnline bool) []int {
 	return InstanceClusterMgr().GetNodeIdByServiceName(serviceName, bOnline)
 }
 
+//获取服务的负载均衡信息
+//负载均衡的策略是从配置获取所有配置了该服务的NodeId 并按NodeId排序 每个node负责处理数组index所在的那一部分
+func GetBalancingInfo(currentNodeId int, servicename string, inSubNet bool) (*BalancingInfo, error) {
+	return InstanceClusterMgr().GetBalancingInfo(currentNodeId, servicename, inSubNet)
+}
+
 //随机选择在线的node发送
 func CallRandomService(NodeServiceMethod string, args interface{}, reply interface{}) error {
 	return InstanceClusterMgr().CallRandomService(NodeServiceMethod, args, reply)

cluster/config.go

@@ -8,6 +8,34 @@ import (
 	"strings"
 )
 
+//负载均衡信息
+type BalancingInfo struct {
+	NodeId      int    //我的nodeId
+	ServiceName string //负载均衡的ServiceName
+
+	TotalNum int   //总共有多少个协同Node
+	MyIndex  int   //负责的index [0, TotalNum)
+	NodeList []int //所有协同的node列表 按NodeId升序排列
+}
+
+//判断hash后的Id是否命中我的NodeId
+func (slf *BalancingInfo) Hit(hashId int) bool {
+	if hashId >= 0 && slf.TotalNum > 0 && slf.MyIndex >= 0 {
+		return hashId%slf.TotalNum == slf.MyIndex
+	}
+	return false
+}
+
+//判断命中的NodeId，-1表示无法取得
+func (slf *BalancingInfo) GetHitNodeId(hashId int) int {
+	if hashId >= 0 && slf.TotalNum > 0 {
+		if idx := hashId % slf.TotalNum; idx >= 0 && idx < len(slf.NodeList) {
+			return slf.NodeList[idx]
+		}
+	}
+	return -1
+}
+
 type CNodeCfg struct {
 	NodeID      int
 	NodeName    string
@@ -148,14 +176,16 @@ func ReadCfg(path string, nodeid int, mapNodeData map[int]NodeData) (*ClusterCon
 	return clsCfg, nil
 }
 
-func (slf *ClusterConfig) GetIdByService(serviceName string) []int {
+func (slf *ClusterConfig) GetIdByService(serviceName, subNetName string) []int {
-	var nodeidlist []int
+	var nodeidlist = []int{}
-	nodeidlist = make([]int, 0)
 
 	nodeList, ok := slf.mapClusterServiceNode[serviceName]
 	if ok == true {
+		nodeidlist = make([]int, 0, len(nodeList))
 		for _, v := range nodeList {
-			nodeidlist = append(nodeidlist, v.NodeID)
+			if subNetName == "" || subNetName == v.SubNetName {
+				nodeidlist = append(nodeidlist, v.NodeID)
+			}
 		}
 	}
 

sysmodule/DBModule.go

@@ -34,6 +34,12 @@ type DBModule struct {
 	syncCoroutineNum int
 }
 
+// Tx ...
+type Tx struct {
+	tx        *sql.Tx
+	PrintTime time.Duration
+}
+
 // DBResult ...
 type DBResult struct {
 	Err          error
@@ -476,7 +482,7 @@ func (slf *DBModule) QueryEx(query string, args ...interface{}) (*DataSetList, e
 		hasRet := rows.NextResultSet()
 
 		if hasRet == false {
-			if rows.Err()!= nil {
+			if rows.Err() != nil {
 				service.GetLogger().Printf(service.LEVER_ERROR, "Query:%s(%+v)", query, rows)
 			}
 			break
@@ -728,3 +734,211 @@ func (slf *DataSetList) rowData2interface(rowIdx int, m map[string][]interface{}
 	}
 	return nil
 }
+
+// Begin starts a transaction.
+func (slf *DBModule) Begin() (*Tx, error) {
+	var txDBMoudule Tx
+	txdb, err := slf.db.Begin()
+	if err != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "Begin error:%s", err.Error())
+		return &txDBMoudule, err
+	}
+	txDBMoudule.tx = txdb
+	return &txDBMoudule, nil
+}
+
+// Rollback aborts the transaction.
+func (slf *Tx) Rollback() error {
+	return slf.tx.Rollback()
+}
+
+// Commit commits the transaction.
+func (slf *Tx) Commit() error {
+	return slf.tx.Commit()
+}
+
+// CheckArgs...
+func (slf *Tx) CheckArgs(args ...interface{}) error {
+	for _, val := range args {
+		if reflect.TypeOf(val).Kind() == reflect.String {
+			retVal := val.(string)
+			if strings.Contains(retVal, "-") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(retVal, "#") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(retVal, "&") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(retVal, "=") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(retVal, "%") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(retVal, "'") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(strings.ToLower(retVal), "delete ") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(strings.ToLower(retVal), "truncate ") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(strings.ToLower(retVal), " or ") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(strings.ToLower(retVal), "from ") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+			if strings.Contains(strings.ToLower(retVal), "set ") == true {
+				return fmt.Errorf("CheckArgs is error arg is %+v", retVal)
+			}
+		}
+	}
+
+	return nil
+}
+
+// Query executes a query that returns rows, typically a SELECT.
+func (slf *Tx) Query(query string, args ...interface{}) DBResult {
+	if slf.CheckArgs(args) != nil {
+		ret := DBResult{}
+		service.GetLogger().Printf(service.LEVER_ERROR, "CheckArgs is error :%s", query)
+		ret.Err = fmt.Errorf("CheckArgs is error!")
+		return ret
+	}
+
+	if slf.tx == nil {
+		ret := DBResult{}
+		service.GetLogger().Printf(service.LEVER_ERROR, "cannot connect database:%s", query)
+		ret.Err = fmt.Errorf("cannot connect database!")
+		return ret
+	}
+
+	rows, err := slf.tx.Query(query, args...)
+	if err != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "Tx Query:%s(%v)", query, err)
+	}
+
+	return DBResult{
+		Err:  err,
+		res:  rows,
+		tag:  "json",
+		blur: true,
+	}
+}
+
+// IsPrintTimeLog...
+func (slf *Tx) IsPrintTimeLog(Time time.Duration) bool {
+	if slf.PrintTime != 0 && Time >= slf.PrintTime {
+		return true
+	}
+	return false
+}
+
+// QueryEx executes a query that return rows.
+func (slf *Tx) QueryEx(query string, args ...interface{}) (*DataSetList, error) {
+	datasetList := DataSetList{}
+	datasetList.tag = "json"
+	datasetList.blur = true
+
+	if slf.CheckArgs(args) != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "CheckArgs is error :%s", query)
+		return &datasetList, fmt.Errorf("CheckArgs is error!")
+	}
+
+	if slf.tx == nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "cannot connect database:%s", query)
+		return &datasetList, fmt.Errorf("cannot connect database!")
+	}
+
+	TimeFuncStart := time.Now()
+	rows, err := slf.tx.Query(query, args...)
+	TimeFuncPass := time.Since(TimeFuncStart)
+
+	if slf.IsPrintTimeLog(TimeFuncPass) {
+		service.GetLogger().Printf(service.LEVER_INFO, "Tx QueryEx Time %s , Query :%s , args :%+v", TimeFuncPass, query, args)
+	}
+	if err != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "Tx Query:%s(%v)", query, err)
+		if rows != nil {
+			rows.Close()
+		}
+		return &datasetList, err
+	}
+	defer rows.Close()
+
+	for {
+		dbResult := DBResultEx{}
+		//取出当前结果集所有行
+		for rows.Next() {
+			if dbResult.RowInfo == nil {
+				dbResult.RowInfo = make(map[string][]interface{})
+			}
+			//RowInfo map[string][][]sql.NullString //map[fieldname][row][column]sql.NullString
+			colField, err := rows.Columns()
+			if err != nil {
+				return &datasetList, err
+			}
+			count := len(colField)
+			valuePtrs := make([]interface{}, count)
+			for i := 0; i < count; i++ {
+				valuePtrs[i] = &sql.NullString{}
+			}
+			rows.Scan(valuePtrs...)
+
+			for idx, fieldname := range colField {
+				fieldRowData := dbResult.RowInfo[strings.ToLower(fieldname)]
+				fieldRowData = append(fieldRowData, valuePtrs[idx])
+				dbResult.RowInfo[strings.ToLower(fieldname)] = fieldRowData
+			}
+			dbResult.rowNum += 1
+		}
+
+		datasetList.dataSetList = append(datasetList.dataSetList, dbResult)
+		//取下一个结果集
+		hasRet := rows.NextResultSet()
+
+		if hasRet == false {
+			if rows.Err() != nil {
+				service.GetLogger().Printf(service.LEVER_ERROR, "Tx Query:%s(%+v)", query, rows)
+			}
+			break
+		}
+	}
+
+	return &datasetList, nil
+}
+
+// Exec executes a query that doesn't return rows.
+func (slf *Tx) Exec(query string, args ...interface{}) (*DBResultEx, error) {
+	ret := &DBResultEx{}
+	if slf.tx == nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "cannot connect database:%s", query)
+		return ret, fmt.Errorf("cannot connect database!")
+	}
+
+	if slf.CheckArgs(args) != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "CheckArgs is error :%s", query)
+		//return ret, fmt.Errorf("cannot connect database!")
+		return ret, fmt.Errorf("CheckArgs is error!")
+	}
+
+	TimeFuncStart := time.Now()
+	res, err := slf.tx.Exec(query, args...)
+	TimeFuncPass := time.Since(TimeFuncStart)
+	if slf.IsPrintTimeLog(TimeFuncPass) {
+		service.GetLogger().Printf(service.LEVER_INFO, "Tx QueryEx Time %s , Query :%s , args :%+v", TimeFuncPass, query, args)
+	}
+	if err != nil {
+		service.GetLogger().Printf(service.LEVER_ERROR, "Tx Exec:%s(%v)", query, err)
+		return nil, err
+	}
+
+	ret.LastInsertID, _ = res.LastInsertId()
+	ret.RowsAffected, _ = res.RowsAffected()
+
+	return ret, nil
+}

sysmodule/DBModule_test.go

@@ -1,6 +1,7 @@
 package sysmodule_test
 
 import (
+	"fmt"
 	"sync"
 	"testing"
 
@@ -13,38 +14,76 @@ func TestDBModule(t *testing.T) {
 	db.ExitChan = make(chan bool)
 	db.WaitGroup = new(sync.WaitGroup)
 
-	db.Init(100, "192.168.0.5:3306", "root", "Root!!2018", "QuantFundsDB")
+	// db.Init(100, "192.168.0.5:3306", "root", "Root!!2018", "QuantFundsDB")
+	db.Init(100, "127.0.0.1:3306", "root", "zgh50221", "rebort_message")
 	db.OnInit()
-	res, err := db.QueryEx("select * from tbl_fun_heelthrow where id >= 1")
+	tx, err := db.GetTx()
 	if err != nil {
-		t.Error(err)
+		fmt.Println("err 1", err)
+		return
+	}
+	res, err := tx.QueryEx("select id as Id, info_type as InfoType, info_type_Name as InfoTypeName from tbl_info_type where id >= 1")
+	if err != nil {
+		fmt.Println("err 2", err)
+		tx.Rollback()
+		return
 	}
 	out := []struct {
-		Addtime int64  `json:"addtime"`
+		Id           int64
-		Tname   string `json:"tname"`
+		InfoType     string
-		Uuid    string `json:"uuid,omitempty"`
+		InfoTypeName string
-		AAAA    string `json:"xxx"`
 	}{}
 	err = res.UnMarshal(&out)
 	if err != nil {
-		t.Error(err)
+		fmt.Println("err 3", err)
+		tx.Rollback()
+		return
 	}
-
+	fmt.Println(out)
-	sres := db.SyncQuery("select * from tbl_fun_heelthrow where id >= 1")
+	_, err = tx.Exec("insert into tbl_info_type(info_type, info_type_name) VALUES (?, ?)", "4", "weibo")
-	res, err = sres.Get(2000)
 	if err != nil {
-		t.Error(err)
+		fmt.Println("err 4", err)
+		tx.Rollback()
+		return
 	}
-
+	_, err = tx.Exec("update tbl_info_type set info_types = ? Where id = ?", "5", 0)
-	out2 := []struct {
-		Addtime int64  `json:"addtime"`
-		Tname   string `json:"tname"`
-		Uuid    string `json:"uuid,omitempty"`
-		AAAA    string `json:"xxx"`
-	}{}
-
-	err = res.UnMarshal(&out2)
 	if err != nil {
-		t.Error(err)
+		fmt.Println("err 4", err)
+		tx.Rollback()
+		return
 	}
+
+	tx.Commit()
+	// res, err := db.QueryEx("select * from tbl_fun_heelthrow where id >= 1")
+	// if err != nil {
+	// 	t.Error(err)
+	// }
+	// out := []struct {
+	// 	Addtime int64  `json:"addtime"`
+	// 	Tname   string `json:"tname"`
+	// 	Uuid    string `json:"uuid,omitempty"`
+	// 	AAAA    string `json:"xxx"`
+	// }{}
+	// err = res.UnMarshal(&out)
+	// if err != nil {
+	// 	t.Error(err)
+	// }
+
+	// sres := db.SyncQuery("select * from tbl_fun_heelthrow where id >= 1")
+	// res, err = sres.Get(2000)
+	// if err != nil {
+	// 	t.Error(err)
+	// }
+
+	// out2 := []struct {
+	// 	Addtime int64  `json:"addtime"`
+	// 	Tname   string `json:"tname"`
+	// 	Uuid    string `json:"uuid,omitempty"`
+	// 	AAAA    string `json:"xxx"`
+	// }{}
+
+	// err = res.UnMarshal(&out2)
+	// if err != nil {
+	// 	t.Error(err)
+	// }
 }

sysmodule/LogModule.go

@@ -55,7 +55,13 @@ func (slf *LogModule) GetCurrentFileName() string {
 	now := time.Now()
 	fpath := filepath.Join("logs")
 	os.MkdirAll(fpath, os.ModePerm)
-	fname := slf.logfilename + "-" + now.Format("20060102-150405") + ".log"
+	y, m, d := now.Date()
+	h := now.Hour()
+	mm := now.Minute()
+	mm -= mm % 15 //15分钟内使用同一个日志文件
+	dt := y*10000 + int(m)*100 + d
+	tm := h*100 + mm
+	fname := fmt.Sprintf("%s-%d-%d.log", slf.logfilename, dt, tm)
 	ret := filepath.Join(fpath, fname)
 	return ret
 }
@@ -84,9 +90,10 @@ func (slf *LogModule) CheckAndGenFile(fileline string) (newFile bool) {
 		}
 
 		var err error
-		slf.logFile, err = os.OpenFile(slf.GetCurrentFileName(), os.O_RDWR|os.O_CREATE|os.O_APPEND, os.ModePerm)
+		filename := slf.GetCurrentFileName()
+		slf.logFile, err = os.OpenFile(filename, os.O_RDWR|os.O_CREATE|os.O_APPEND, os.ModePerm)
 		if err != nil {
-			fmt.Printf("create log file %+v error!", slf.GetCurrentFileName())
+			fmt.Printf("create log file %+v error!", filename)
 			slf.locker.Unlock()
 			return false
 		}