Go语言中如何高效处理多数据库动态连接？

核心思路 #

1. 连接池缓存：按需创建并缓存数据库连接池，避免重复建立连接消耗资源
2. 请求级连接注入：通过Echo中间件从查询参数提取目标数据库名，将对应连接池注入请求上下文
3. 适配分层架构：修改Repository和Usecase层，支持从上下文动态获取数据库连接

代码修改实现 #

1. 新增连接池管理器（Main.go扩展） #

定义管理器负责连接池的创建、缓存和销毁：

package main

import (
    "database/sql"
    "fmt"
    "log"
    "sync"
    "time"

    _ "github.com/go-sql-driver/mysql"
    _ "github.com/lib/pq"

    "github.com/labstack/echo"
    "github.com/spf13/viper"

    _variantHttpDelivery "backend/server/variant/delivery/http"
    _variantHttpDeliveryMiddleware "backend/server/variant/delivery/http/middleware"
    _variantUcase "backend/server/variant/usecase"
)

// DBPoolManager 管理多个数据库连接池
type DBPoolManager struct {
    sync.RWMutex
    pools   map[string]*sql.DB
    dbHost  string
    dbPort  string
    dbUser  string
    dbPass  string
}

// NewDBPoolManager 初始化连接池管理器
func NewDBPoolManager(host, port, user, pass string) *DBPoolManager {
    return &DBPoolManager{
        pools:  make(map[string]*sql.DB),
        dbHost: host,
        dbPort: port,
        dbUser: user,
        dbPass: pass,
    }
}

// GetPool 根据数据库名获取连接池，不存在则创建
func (m *DBPoolManager) GetPool(dbName string) (*sql.DB, error) {
    m.RLock()
    pool, exists := m.pools[dbName]
    m.RUnlock()

    if exists {
        return pool, nil
    }

    m.Lock()
    defer m.Unlock()

    // 双重检查防止并发重复创建
    if pool, exists = m.pools[dbName]; exists {
        return pool, nil
    }

    connection := fmt.Sprintf("postgresql://%s:%s@%s:%s/%s", m.dbUser, m.dbPass, m.dbHost, m.dbPort, dbName)
    dsn := fmt.Sprintf("%s?sslmode=disable", connection)
    newPool, err := sql.Open(`postgres`, dsn)
    if err != nil {
        return nil, err
    }

    // 连接池参数配置（最佳实践）
    newPool.SetMaxOpenConns(20)
    newPool.SetMaxIdleConns(10)
    newPool.SetConnMaxLifetime(1 * time.Hour)
    newPool.SetConnMaxIdleTime(30 * time.Minute)

    // 验证连接有效性
    if err := newPool.Ping(); err != nil {
        _ = newPool.Close()
        return nil, err
    }

    log.Printf("成功连接数据库: %s 👍", dbName)
    m.pools[dbName] = newPool
    return newPool, nil
}

// CloseAll 关闭所有连接池
func (m *DBPoolManager) CloseAll() error {
    m.Lock()
    defer m.Unlock()

    var err error
    for name, pool := range m.pools {
        if closeErr := pool.Close(); closeErr != nil {
            log.Printf("关闭数据库%s连接池失败: %v", name, closeErr)
            err = closeErr
        }
    }
    return err
}

var dbManager *DBPoolManager

func init() {
    viper.SetConfigFile(`config.json`)
    err := viper.ReadInConfig()
    if err != nil {
        panic(err)
    }

    if viper.GetBool(`debug`) {
        log.Println("服务运行在DEBUG模式")
    }
}

func main() {
    dbHost := viper.GetString(`database.host`)
    dbPort := viper.GetString(`database.port`)
    dbUser := viper.GetString(`database.user`)
    dbPass := viper.GetString(`database.pass`)

    dbManager = NewDBPoolManager(dbHost, dbPort, dbUser, dbPass)

    // 优雅关闭所有连接池
    defer func() {
        if err := dbManager.CloseAll(); err != nil {
            log.Fatal(err)
        }
    }()

    e := echo.New()
    middL := _variantHttpDeliveryMiddleware.InitMiddleware(dbManager)
    e.Use(middL.CORS)
    e.Use(middL.DynamicDBMiddleware) // 注入动态数据库中间件

    timeoutContext := time.Duration(viper.GetInt("context.timeout")) * time.Second
    au := _variantUcase.NewVariantUsecase(timeoutContext)
    _variantHttpDelivery.NewVariantHandler(e, au)

    log.Fatal(e.Start(viper.GetString("server.address"))) //nolint
}

2. 新增动态数据库中间件（middleware层） #

从查询参数提取数据库名，校验合法性后注入上下文：

package middleware

import (
    "net/http"
    "strings"

    "github.com/labstack/echo"
    "backend/server/main" // 替换为实际main包路径
)

type Middleware struct {
    dbManager *main.DBPoolManager
}

func InitMiddleware(dbManager *main.DBPoolManager) *Middleware {
    return &Middleware{dbManager: dbManager}
}

// DynamicDBMiddleware 处理动态数据库连接逻辑
func (m *Middleware) DynamicDBMiddleware(next echo.HandlerFunc) echo.HandlerFunc {
    return func(c echo.Context) error {
        dbName := c.QueryParam("db")
        if dbName == "" {
            return c.JSON(http.StatusBadRequest, map[string]string{"error": "缺少'db'查询参数"})
        }

        // 数据库名称白名单校验，防止非法访问
        allowedDBs := []string{"db1", "db2", "db3", "..."} // 列出所有22个数据库名
        isValid := false
        for _, allowed := range allowedDBs {
            if strings.EqualFold(dbName, allowed) {
                isValid = true
                break
            }
        }
        if !isValid {
            return c.JSON(http.StatusBadRequest, map[string]string{"error": "非法数据库名称"})
        }

        pool, err := m.dbManager.GetPool(dbName)
        if err != nil {
            return c.JSON(http.StatusInternalServerError, map[string]string{"error": "数据库连接失败"})
        }

        c.Set("db-conn", pool)
        return next(c)
    }
}

// 保留原有CORS中间件逻辑
func (m *Middleware) CORS(next echo.HandlerFunc) echo.HandlerFunc {
    return func(c echo.Context) error {
        // 原有CORS代码不变
        return next(c)
    }
}

3. 修改Repository层，支持动态获取连接 #

移除固定连接持有，从上下文获取连接池：

package postgres

import (
    "backend/server/domain"
    "context"
    "database/sql"
    "github.com/sirupsen/logrus"
    "reflect"

    "github.com/labstack/echo"
)

type psqlVariantRepository struct{}

func NewPsqlVariantRepository() domain.VariantRepository {
    return &psqlVariantRepository{}
}

func (m *psqlVariantRepository) GetByVCF(ctx context.Context, vcf string) (res domain.Variant, err error) {
    ec, ok := ctx.(echo.Context)
    if !ok {
        return domain.Variant{}, domain.ErrInvalidContext
    }

    conn, ok := ec.Get("db-conn").(*sql.DB)
    if !ok {
        return domain.Variant{}, domain.ErrDBConnectionNotFound
    }

    query := `SELECT * FROM main1 WHERE variant_vcf = $1`
    list, err := m.fetch(ctx, conn, query, vcf)

    if err != nil {
        return domain.Variant{}, err
    }

    if len(list) > 0 {
        res = list[0]
    } else {
        return res, domain.ErrNotFound
    }
    return
}

// fetch 修改为接收连接参数
func (m *psqlVariantRepository) fetch(ctx context.Context, conn *sql.DB, query string, args ...interface{}) (result []domain.Variant, err error) {
    rows, err := conn.QueryContext(ctx, query, args...)
    if err != nil {
        logrus.Error(err)
        return nil, err
    }

    defer func() {
        if errRow := rows.Close(); errRow != nil {
            logrus.Error(errRow)
        }
    }()

    result = make([]domain.Variant, 0)
    for rows.Next() {
        t := domain.Variant{}
        values := make([]interface{}, 0, reflect.TypeOf(t).NumField())
        v := reflect.ValueOf(&t).Elem()

        for i := 0; i < v.NumField(); i++ {
            if v.Type().Field(i).Type.Kind() == reflect.String {
                values = append(values, new(sql.NullString))
            } else {
                values = append(values, v.Field(i).Addr().Interface())
            }
        }

        if err = rows.Scan(values...); err != nil {
            logrus.Error(err)
            return nil, err
        }

        for i, value := range values {
            if ns, ok := value.(*sql.NullString); ok {
                v.Field(i).SetString(ns.String)
            }
        }

        result = append(result, t)
    }

    logrus.Info("成功从数据库获取结果 👍")
    return result, nil
}

4. 修改Usecase层适配动态连接 #

调整Usecase初始化逻辑，不再依赖固定Repository连接：

package usecase

import (
    "backend/server/domain"
    "backend/server/variant/repository/postgres"
    "context"
    "time"
)

type variantUsecase struct {
    variantRepo domain.VariantRepository
    timeout     time.Duration
}

func NewVariantUsecase(timeout time.Duration) domain.VariantUsecase {
    return &variantUsecase{
        variantRepo: postgres.NewPsqlVariantRepository(),
        timeout:     timeout,
    }
}

func (u *variantUsecase) GetByVCF(ctx context.Context, vcf string) (domain.Variant, error) {
    ctx, cancel := context.WithTimeout(ctx, u.timeout)
    defer cancel()

    return u.variantRepo.GetByVCF(ctx, vcf)
}

最佳实践 #

1. 连接池参数调优：根据业务QPS调整SetMaxOpenConns、SetMaxIdleConns等参数，避免连接泄漏或资源浪费
2. 严格参数校验：必须对查询参数中的数据库名做白名单校验，防止非法访问
3. 连接池缓存复用：通过带锁的map缓存连接池，减少重复建立连接的开销
4. 优雅资源释放：程序退出时关闭所有连接池，避免资源泄漏
5. 监控与日志：添加连接池状态监控（如活跃连接数、空闲连接数），记录关键操作日志便于排查问题
6. 请求隔离：通过上下文传递连接池，避免全局变量滥用，保证请求间资源隔离

内容的提问来源于stack exchange，提问作者mad