zyos_go_mq_consume/app/db/dbs.go

package db

import (
	"applet/app/db/model"
	"applet/app/utils/logx"
	"fmt"
	"os"
	"time"

	"xorm.io/xorm"
	"xorm.io/xorm/log"

	"applet/app/cfg"
)

var DBs map[string]*xorm.Engine

// 每个站长都要有自己的syscfg 缓存， 键是站长id，值是缓存名
// var SysCfgMapKey map[string]string

func NewDB(c *cfg.DBCfg) (*xorm.Engine, error) {
	db, err := xorm.NewEngine("mysql", fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4", c.User, c.Psw, c.Host, c.Name))
	if err != nil {
		return nil, err
	}
	db.SetConnMaxLifetime(c.MaxLifetime * time.Second)
	db.SetMaxOpenConns(c.MaxOpenConns)
	db.SetMaxIdleConns(c.MaxIdleConns)
	err = db.Ping()
	if err != nil {
		return nil, err
	}
	if c.ShowLog {
		db.ShowSQL(true)
		db.Logger().SetLevel(log.LOG_DEBUG)
		f, err := os.OpenFile(c.Path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0777)
		if err != nil {
			os.RemoveAll(c.Path)
			if f, err = os.OpenFile(c.Path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0777); err != nil {
				return nil, err
			}
		}
		logger := log.NewSimpleLogger(f)
		logger.ShowSQL(true)
		db.SetLogger(logger)
	}
	return db, nil
}

// InitDBs is 初始化多数据库
func InitDBs(ch chan int) {
	// 初始化多数据库
	var tables *[]model.DbMapping
	InitMapDbs(cfg.DB, cfg.Prd)
	ch <- 1
	// 每10s 查询一次模板数据库的db mapping 表，如果有新增数据库记录，则添加到 DBs中
	ticker := time.NewTicker(time.Duration(time.Second * 120))
	for range ticker.C {
		if cfg.Prd {
			tables = GetAllDatabasePrd() //默认获取全部
		} else {
			tables = GetAllDatabaseDev() //默认获取全部
		}
		if tables == nil {
			logx.Warn("no database tables data")
			continue
		}
		for _, item := range *tables {
			_, ok := DBs[item.DbMasterId]
			if !ok {
				// 不在db.DBs 则添加进去
				dbCfg := cfg.DBCfg{
					Name:         item.DbName,
					ShowLog:      cfg.DB.ShowLog,
					MaxLifetime:  cfg.DB.MaxLifetime,
					MaxOpenConns: cfg.DB.MaxOpenConns,
					MaxIdleConns: cfg.DB.MaxIdleConns,
					Path:         fmt.Sprintf(cfg.DB.Path, item.DbName),
				}
				if item.DbHost != "" && item.DbUsername != "" && item.DbPassword != "" {
					dbCfg.Host = item.DbHost
					dbCfg.User = item.DbUsername
					dbCfg.Psw = item.DbPassword
				} else {
					dbCfg.Host = cfg.DB.Host
					dbCfg.User = cfg.DB.User
					dbCfg.Psw = cfg.DB.Psw
				}
				e, err := NewDB(&dbCfg)
				if err != nil || e == nil {
					logx.Warnf("db engine can't create, please check config, params[host:%s, user:%s, psw: %s, name: %s], err: %v", dbCfg.Host, dbCfg.User, dbCfg.Psw, dbCfg.Name, err)
				} else {
					DBs[item.DbMasterId] = e
				}
			}
			// 如果 被禁用则删除
			if item.DeletedAt == 1 {
				logx.Infof("%s have been removed", item.DbMasterId)
				delete(DBs, item.DbMasterId)
			}
		}
	}

}