package database import ( "fmt" "log" "sync" "git.x2erp.com/qdy/go-base/config" "git.x2erp.com/qdy/go-base/ctx" "git.x2erp.com/qdy/go-base/types" "git.x2erp.com/qdy/go-db/drivers" "git.x2erp.com/qdy/go-db/functions" "github.com/jmoiron/sqlx" ) type DBFactory struct { db *sqlx.DB name string // 记录数据库配置名称 } var ( // 多实例存储:配置名称 -> DBFactory 实例 instances = make(map[string]*DBFactory) // 每个配置名称对应的once,确保线程安全 onceMap = make(map[string]*sync.Once) // 保护instances和onceMap的读写锁 instancesMutex sync.RWMutex ) // GetDBFactory 获取指定名称的数据库工厂单例 func GetDBFactory(dbName string) (*DBFactory, error) { // 获取或创建该名称的once对象 instancesMutex.Lock() once, exists := onceMap[dbName] if !exists { once = &sync.Once{} onceMap[dbName] = once } instancesMutex.Unlock() var initErr error var instance *DBFactory var msg = fmt.Sprintf("DBFactory '%s' instance retrieved from memory.\n", dbName) once.Do(func() { // 使用配置单例 cfg, err := config.GetConfig() if err != nil { initErr = fmt.Errorf("failed to load config: %v", err) return } // 获取指定名称的数据库配置 dbConfig := cfg.GetDatabaseConfig(dbName) if dbConfig == nil { initErr = fmt.Errorf("database configuration '%s' not found", dbName) return } // // 检查数据库配置是否完整 // if !dbConfig.IsConfigured() { // initErr = fmt.Errorf("database configuration '%s' is incomplete", dbName) // return // } // 获取数据库类型 dbType := dbConfig.Type log.Printf("Creating database connection for '%s' with type: %s\n", dbName, dbType) // 获取对应的驱动 dbDriver, err := drivers.Get(dbType) if err != nil { initErr = fmt.Errorf("failed to get database driver: %v", err) return } // 将内部 DBConfig 转换为 drivers.DBConfig driverConfig := drivers.DBConfig{ Type: dbConfig.Type, Host: dbConfig.Host, Port: dbConfig.Port, Username: dbConfig.Username, Password: dbConfig.Password, Database: dbConfig.Database, MaxOpenConns: dbConfig.MaxOpenConns, MaxIdleConns: dbConfig.MaxIdleConns, ConnMaxLifetime: dbConfig.ConnMaxLifetime, } // 创建数据库连接 db, err := dbDriver.Open(driverConfig) if err != nil { initErr = fmt.Errorf("failed to open database connection for '%s': %v", dbName, err) return } // 测试连接 if err := functions.TestConnection(db, dbType); err != nil { db.Close() initErr = fmt.Errorf("database connection test failed for '%s': %v", dbName, err) return } msg = fmt.Sprintf("DBFactory '%s' is successfully created.\n", dbName) instance = &DBFactory{ db: db, name: dbName, } // 保存实例到map instancesMutex.Lock() instances[dbName] = instance instancesMutex.Unlock() }) if initErr != nil { return nil, initErr } log.Print(msg) // 从map中获取实例 instancesMutex.RLock() instance = instances[dbName] instancesMutex.RUnlock() return instance, nil } // GetDefaultDBFactory 获取默认数据库工厂(向后兼容) func GetDefaultDBFactory() (*DBFactory, error) { return GetDBFactory("default") } // GetAllDBFactories 获取所有已创建的数据库工厂实例 func GetAllDBFactories() map[string]*DBFactory { instancesMutex.RLock() defer instancesMutex.RUnlock() // 创建副本,避免外部修改 result := make(map[string]*DBFactory) for k, v := range instances { result[k] = v } return result } // GetDBFactoryNames 获取所有可用的数据库配置名称 func GetDBFactoryNames() []string { cfg, err := config.GetConfig() if err != nil { return []string{} } dbs := cfg.GetDatabases() if dbs == nil { return []string{} } return dbs.GetAllDatabaseNames() } // CloseInstance 关闭指定名称的数据库连接 func CloseInstance(dbName string) error { instancesMutex.Lock() defer instancesMutex.Unlock() if instance, exists := instances[dbName]; exists { err := instance.Close() delete(instances, dbName) delete(onceMap, dbName) return err } return fmt.Errorf("database instance '%s' not found", dbName) } // CloseAll 关闭所有数据库连接 func CloseAll() { instancesMutex.Lock() defer instancesMutex.Unlock() for name, instance := range instances { if err := instance.Close(); err != nil { log.Printf("Error closing database instance '%s': %v\n", name, err) } delete(instances, name) delete(onceMap, name) } // 重新初始化maps instances = make(map[string]*DBFactory) onceMap = make(map[string]*sync.Once) log.Println("All database connections closed gracefully") } // ========== DBFactory 实例方法 ========== // GetDB 获取数据库连接(线程安全) func (f *DBFactory) GetDB() interface{} { return f.db } // Close 关闭数据库连接 func (f *DBFactory) Close() error { if f.db != nil { err := f.db.Close() f.db = nil log.Printf("Database connection '%s' closed gracefully\n", f.name) return err } return nil } // GetDBType 得到当前使用数据库类型 func (f *DBFactory) GetDBType() string { // 通过配置获取当前数据库的类型 cfg, err := config.GetConfig() if err != nil { return "" } dbConfig := cfg.GetDatabaseConfig(f.name) if dbConfig == nil { return "" } return dbConfig.Type } // GetDBName 获取数据库配置名称 func (f *DBFactory) GetDBName() string { return f.name } // QueryToJSON 快捷查询,直接返回 JSON 字节流 func (f *DBFactory) QueryToJSON(sql string, reqCtx *ctx.RequestContext) *types.QueryResult[[]map[string]interface{}] { return functions.QueryToJSON(f.db, sql, reqCtx) } // QueryParamsToJSON 位置参数查询并返回 JSON 字节数据 func (f *DBFactory) QueryPositionalToJSON(sql string, params []interface{}, reqCtx *ctx.RequestContext) *types.QueryResult[[]map[string]interface{}] { return functions.QueryPositionalToJSON(f.db, sql, params, reqCtx) } // QueryParamsNameToJSON 命名参数查询并返回 JSON 字节数据 // params 可以是 map[string]interface{} 或结构体 func (f *DBFactory) QueryParamsNameToJSON(sql string, params map[string]interface{}, reqCtx *ctx.RequestContext) *types.QueryResult[[]map[string]interface{}] { return functions.QueryParamsNameToJSON(f.db, sql, params, reqCtx) } // QueryToCSV 快捷查询,直接返回 CSV 字符串(包含表头) func (f *DBFactory) QueryToCSV(sql string, writerHeader bool, reqCtx *ctx.RequestContext) ([]byte, error) { return functions.QueryToCSV(f.db, sql, writerHeader, reqCtx) } // QueryParamsToCSV 位置参数查询并返回 CSV 字节数据 func (f *DBFactory) QueryPositionalToCSV(sql string, writerHeader bool, params []interface{}, reqCtx *ctx.RequestContext) ([]byte, error) { return functions.QueryPositionalToCSV(f.db, sql, writerHeader, params, reqCtx) } // QueryParamsNameToCSV 命名参数查询并返回 CSV 字节数据 // params 可以是 map[string]interface{} 或结构体 func (f *DBFactory) QueryParamsNameToCSV(sql string, writerHeader bool, params map[string]interface{}, reqCtx *ctx.RequestContext) ([]byte, error) { return functions.QueryParamsNameToCSV(f.db, sql, writerHeader, params, reqCtx) } // ExecuteDDL 快捷执行DDL语句 func (f *DBFactory) ExecuteDDL(ddlSQL string) error { return functions.ExecuteDDL(f.db, ddlSQL) } // ExecuteDDLWithTx 快捷在事务中执行DDL语句 func (f *DBFactory) ExecuteDDLWithTx(ddlSQL string) error { return functions.ExecuteDDLWithTx(f.db, ddlSQL) } // ExecuteMultipleDDL 快捷执行多个DDL语句 func (f *DBFactory) ExecuteMultipleDDL(ddlSQLs []string) error { return functions.ExecuteMultipleDDL(f.db, ddlSQLs) } // GetAvailableDrivers 获取可用的数据库驱动 func (f *DBFactory) GetAvailableDrivers() []string { return drivers.GetAllDrivers() } func (f *DBFactory) TestConnection(dbType string) error { return functions.TestConnection(f.db, dbType) }