golang源码分析：gorm

golangLeetcode

发布于 2022-08-03 14:00:10

2.3K0

发布于 2022-08-03 14:00:10

文章被收录于专栏：golang算法架构leetcode技术phpgolang算法架构leetcode技术php

gorm使用很简洁，首先打开数据库连接（Open initialize a new db connection, need to import driver first）

db, err = gorm.Open(mysql.Open(dbDSN), &gorm.Config{})

进行连接初始化的方法Open函数定义在github.com/go-gorm/gorm/gorm.go：

func Open(dialector Dialector, opts ...Option) (db *DB, err error) {
 // 1.初始化配置，通过opts 来设置可变参数
  config := &Config{}
  // 2.配置进行应用
  if d, ok := dialector.(interface{ Apply(*Config) error }); ok {
    if err = d.Apply(config); err != nil {
      return
    }
  }
  // 3.初始化gorm.DB对象，后续操作通过clone 该对象进行调用
  db = &DB{Config: config, clone: 1}
  // 初始化执行函数
  db.callbacks = initializeCallbacks(db)
  db.Statement = &Statement{
    DB:       db,
    ConnPool: db.ConnPool,
    Context:  context.Background(),
    Clauses:  map[string]clause.Clause{},
  }
  // 4.通过Initialize方法建立连接
  if dialector != nil {
    config.Dialector = dialector
  }
  if config.Dialector != nil {
    err = config.Dialector.Initialize(db)
  }
}

我们常用的github.com/jinzhu/gorm是从官方fork过来的，进行了一系列的封装，比如连接可以直接使用dsn

 db, err := gorm.Open("mysql", "user:password@/dbname?charset=utf8&parseTime=True&loc=Local")

它的Open函数支持多种数据库：

func Open(dialect string, args ...interface{}) (db *DB, err error) {
   dbSQL, err = sql.Open(driver, source)
   db = &DB{
    db:        dbSQL,
    logger:    defaultLogger,
    callbacks: DefaultCallback,
    dialect:   newDialect(dialect, dbSQL),
  }
  db.parent = db
}

那么两者的区别有哪些呢？我们用的时候应该如何抉择？

1，连接方式不同

// jinzhu
func Open(dialect string, args ...interface{}) (db *DB, err error) {}

// grom.io
func Open(dialector Dialector, opts ...Option) (db *DB, err error) {}

2，查询结果空值处理方式不一样

gorm.io 的 Find 函数在进行查找时，如果查找结果为空，不会报record not found，当接收函数为集合时，返回空集合；非集合时，返回零值

3，更新的方式不一样

Jinzhu 版本支持传参为结构体，但结构体为零值时 sql 不执行

gorm.io 版本必须传两个参数，传结构体用Updates

4，where条件不一致

jinzhu版在调用 Where 时会创建一个副本，同一个 DB 在多行调用 Where 函数时内容不会叠加

gormio版同一个 DB 在多行调用 Where 函数时内容会叠加

下面我们以jinzhu/orm版本为例来分析源码。初始化连接后可以开始使用：

// 根据主键查询第一条记录
db.First(&user)
//// SELECT * FROM users ORDER BY id LIMIT 1;

// 随机获取一条记录
db.Take(&user)
//// SELECT * FROM users LIMIT 1;

// 根据主键查询最后一条记录
db.Last(&user)
//// SELECT * FROM users ORDER BY id DESC LIMIT 1;

// 查询所有的记录
db.Find(&users)
//// SELECT * FROM users;

// 查询指定的某条记录(仅当主键为整型时可用)
db.First(&user, 10)
// SELECT * FROM users WHERE id = 10;
// First find first record that match given conditions, order by primary key

gorm使用builder模式将SQL各种表达通过实现Build方法来生成对应字符串。Builder模式的定义是“将一个复杂对象的构建与它的表示分离，使得同样的构建过程可以创建不同的表示。”它属于创建类模式，一个对象的构建比较复杂，超出了构造函数所能包含的范围，就可以使用工厂模式和Builder模式，相对于工厂模式会产出一个完整的产品，Builder应用于更加复杂的对象的构建，甚至只会构建产品的一个部分。Factory模式一进一出,Builder模式是分步流水线作业。当你需要做一系列有序的工作或者按照一定的逻辑来完成创建一个对象时 Builder就派上用场。正好适合我们的场景。可分为两个阶段：存储数据+处理数据；GORM的调用就是采用了chainable+finisher的两段实现，前者保存SQL相关元数据，后者拼接SQL并执行；

我们以First函数为例进行研究github.com/jinzhu/gorm/main.go

// First find first record that match given conditions, order by primary key
func (s *DB) First(out interface{}, where ...interface{}) *DB {
  newScope := s.NewScope(out)
  newScope.Search.Limit(1)
  return newScope.Set("gorm:order_by_primary_key", "ASC").
    inlineCondition(where...)
    .callCallbacks(s.parent.callbacks.queries).db
}

它首先创建了一个新的scope

// NewScope create a scope for current operation
func (s *DB) NewScope(value interface{}) *Scope {
  dbClone := s.clone()
  dbClone.Value = value
  return &Scope{db: dbClone, Search: dbClone.search.clone(), Value: value}
}

克隆了一个db实例，设置输出值，相当于开辟了一个干净的数据库“交互环境”。这个克隆的db实例，包裹在Scope里面。在刚才First方法里面，也就是First方法内有效。所以，业务代码持有的总是最原始的db实例，即通过gorm.Open出来的db实例。

然后使用构建者模式分多步构建出我们的db实例：callCallback是逐步对多个Callback发起call，也就是按顺序调用callbacks。代码位于

github.com/jinzhu/gorm/scope.go：

func (scope *Scope) inlineCondition(values ...interface{}) *Scope {
  if len(values) > 0 {
    scope.Search.Where(values[0], values[1:]...)
  }
  return scope
}

func (scope *Scope) callCallbacks(funcs []*func(s *Scope)) *Scope {
  defer func() {
    if err := recover(); err != nil {
      if db, ok := scope.db.db.(sqlTx); ok {
        db.Rollback()
      }
      panic(err)
    }
  }()
  for _, f := range funcs {
    (*f)(scope)
    if scope.skipLeft {
      break
    }
  }
  return scope
}

依次调用，每一个没有被跳过的callback

每个Callback做一件事情，比如读取数据库值mapping到struct，级联读取其他值。这样好处是：1，callback设计简单，做一件事2，拓展性好，即s.parent.callbacks.queries, s.parent.callbacks.queries, s.parent.callbacks.deletes等执行过程可随意扩展。通过Open函数，我们可知callbacks是通过DefaultCallback来进行赋值的。那么DefaultCallback是如何初始化的呢？

github.com/jinzhu/gorm/callback.go

var DefaultCallback = &Callback{}

它定义了一系列callback

type Callback struct {
  logger     logger
  creates    []*func(scope *Scope)
  updates    []*func(scope *Scope)
  deletes    []*func(scope *Scope)
  queries    []*func(scope *Scope)
  rowQueries []*func(scope *Scope)
  processors []*CallbackProcessor
}

这些callback是在什么对方进行初始化的呢？在各个callback_*.go的init方法，比如callback_query.go：

func init() {
  DefaultCallback.Query().Register("gorm:query", queryCallback)
  DefaultCallback.Query().Register("gorm:preload", preloadCallback)
  DefaultCallback.Query().Register("gorm:after_query", afterQueryCallback)
}

其中Query()定义如下

func (c *Callback) Query() *CallbackProcessor {
  return &CallbackProcessor{logger: c.logger, kind: "query", parent: c}
}

type CallbackProcessor struct {
  logger    logger
  name      string              // current callback's name
  before    string              // register current callback before a callback
  after     string              // register current callback after a callback
  replace   bool                // replace callbacks with same name
  remove    bool                // delete callbacks with same name
  kind      string              // callback type: create, update, delete, query, row_query
  processor *func(scope *Scope) // callback handler
  parent    *Callback
}

// Register a new callback, refer `Callbacks.Create`
func (cp *CallbackProcessor) Register(callbackName string, callback func(scope *Scope)) {
  if cp.kind == "row_query" {
    if cp.before == "" && cp.after == "" && callbackName != "gorm:row_query" {
      cp.logger.Print(fmt.Sprintf("Registering RowQuery callback %v without specify order with Before(), After(), applying Before('gorm:row_query') by default for compatibility...\n", callbackName))
      cp.before = "gorm:row_query"
    }
  }

  cp.name = callbackName
  cp.processor = &callback
  cp.parent.processors = append(cp.parent.processors, cp)
  cp.parent.reorder()
}

下面我们看下queryCallback具体做了什么

// queryCallback used to query data from database
func queryCallback(scope *Scope) {
    if _, skip := scope.InstanceGet("gorm:skip_query_callback"); skip {
        return
    }
    //we are only preloading relations, dont touch base model
    if _, skip := scope.InstanceGet("gorm:only_preload"); skip {
        return
    }
    defer scope.trace(scope.db.nowFunc())
    var (
        isSlice, isPtr bool
        resultType     reflect.Type
        results        = scope.IndirectValue()
    )

    if orderBy, ok := scope.Get("gorm:order_by_primary_key"); ok {
        if primaryField := scope.PrimaryField(); primaryField != nil {
            scope.Search.Order(fmt.Sprintf("%v.%v %v", scope.QuotedTableName(), scope.Quote(primaryField.DBName), orderBy))
        }
    }
    
    if kind := results.Kind(); kind == reflect.Slice {
    isSlice = true
    resultType = results.Type().Elem()
    results.Set(reflect.MakeSlice(results.Type(), 0, 0))

    if resultType.Kind() == reflect.Ptr {
      isPtr = true
      resultType = resultType.Elem()
    }
  } else if kind != reflect.Struct {
    scope.Err(errors.New("unsupported destination, should be slice or struct"))
    return
  }

  scope.prepareQuerySQL()
  if rows, err := scope.SQLDB().Query(scope.SQL, scope.SQLVars...); scope.Err(err) == nil {
     scope.scan(rows, columns, scope.New(elem.Addr().Interface()).Fields())
  }

核心的步骤在于 scope.prepareQuerySQL() 构建 SQL 语句.

func (scope *Scope) prepareQuerySQL() {
    if scope.Search.raw {
        scope.Raw(scope.CombinedConditionSql())
    } else {
        scope.Raw(fmt.Sprintf("SELECT %v FROM %v %v", scope.selectSQL(), scope.QuotedTableName(), scope.CombinedConditionSql()))
    }
    return
}

具体的SELECT 表达式需要三个变量, 字段名, 表名, 条件.

func (scope *Scope) selectSQL() string {
    if len(scope.Search.selects) == 0 {
        if len(scope.Search.joinConditions) > 0 {
            return fmt.Sprintf("%v.*", scope.QuotedTableName())
        }
        return "*"
    }
    return scope.buildSelectQuery(scope.Search.selects)
}

func (scope *Scope) buildSelectQuery(clause map[string]interface{}) (str string) {
    switch value := clause["query"].(type) {
    case string:
        str = value
    case []string:
        str = strings.Join(value, ", ")

func (scope *Scope) QuotedTableName() (name string) {
    if scope.search != nil && len(scope.Search.tableName) > 0 {
        if strings.Contains(scope.Search.tableName, " ") {
            return scope.Search.tableName
        }
        return scope.Quote(scope.Search.tableName)
    }
    return scope.Quote(scope.TableName())
}

// CombinedConditionSql return combined condition sql
func (scope *Scope) CombinedConditionSql() string {
    joinSQL := scope.joinsSQL()
    whereSQL := scope.whereSQL()
    if scope.Search.raw {
        whereSQL = strings.TrimSuffix(strings.TrimPrefix(whereSQL, "WHERE ("), ")")
    }
    return joinSQL + whereSQL + scope.groupSQL() +
        scope.havingSQL() + scope.orderSQL() + scope.limitAndOffsetSQL()
}

func (scope *Scope) buildCondition(clause map[string]interface{}, include bool) (str string) {

然后通过 rows, err := scope.SQLDB().Query(scope.SQL, scope.SQLVars...), 执行了数据库查询.SQL 各种表达通过实现Build方法来生成对应字符串。

func Query(db *gorm.DB) {
  if db.Error == nil {
    // 1.构建查询的SQL
    BuildQuerySQL(db)
    // 2.真正对语句进行执行，并返回对应的Rows结果
    if !db.DryRun && db.Error == nil {
      rows, err := db.Statement.ConnPool.QueryContext(db.Statement.Context, db.Statement.SQL.String(), db.Statement.Vars...)
      gorm.Scan(rows, db, 0)
    }
  }
}

最后调用scan方法完成结果到对象的映射scope.scan(rows, columns, scope.New(elem.Addr().Interface()).Fields()) ，可以看到其中使用了大量的反射。

func (scope *Scope) scan(rows *sql.Rows, columns []string, fields []*Field) {
     for index, column := range columns {
        selectFields = fields
        offset := 0
        if idx, ok := selectedColumnsMap[column]; ok {
            offset = idx + 1
            selectFields = selectFields[offset:]
        }
        for fieldIndex, field := range selectFields {
              if field.DBName == column {
                if field.Field.Kind() == reflect.Ptr {
                    values[index] = field.Field.Addr().Interface()
                } else {
                    reflectValue := reflect.New(reflect.PtrTo(field.Struct.Type))
                    reflectValue.Elem().Set(field.Field.Addr())
                    values[index] = reflectValue.Interface()
                    resetFields[index] = field
                }

类似的我们可以看看删除操作：

db.Delete(&people)

func (s *DB) Delete(value interface{}, where ...interface{}) *DB {
  return s.NewScope(value).inlineCondition(where...).callCallbacks(s.parent.callbacks.deletes).db
}

func (scope *Scope) inlineCondition(values ...interface{}) *Scope {
  if len(values) > 0 {
    scope.Search.Where(values[0], values[1:]...)
  }
  return scope
}

func (s *DB) Callback() *Callback {

  s.parent.callbacks = s.parent.callbacks.clone(s.logger)
  return s.parent.callbacks
}

创建操作

func (c *Callback) Create() *CallbackProcessor {
  return &CallbackProcessor{logger: c.logger, kind: "create", parent: c}
}

github.com/jinzhu/gorm/callback_create.go

func init() {
  DefaultCallback.Create().Register("gorm:begin_transaction", beginTransactionCallback)
  DefaultCallback.Create().Register("gorm:before_create", beforeCreateCallback)
  DefaultCallback.Create().Register("gorm:create", createCallback)

func createCallback(scope *Scope) {
   for _, field := range scope.Fields() {
      if scope.changeableField(field) {
            blankColumnsWithDefaultValue = append(blankColumnsWithDefaultValue, scope.Quote(field.DBName))
            scope.InstanceSet("gorm:blank_columns_with_default_value", blankColumnsWithDefaultValue)
          scope.Raw(fmt.Sprintf(
        "INSERT%v INTO %v (%v)%v VALUES (%v)%v%v",
        addExtraSpaceIfExist(insertModifier),
        scope.QuotedTableName(),
        strings.Join(columns, ","),
        addExtraSpaceIfExist(lastInsertIDOutputInterstitial),
        strings.Join(placeholders, ","),
        addExtraSpaceIfExist(extraOption),
        addExtraSpaceIfExist(lastInsertIDReturningSuffix),
      ))
     if err := scope.SQLDB().QueryRow(scope.SQL, scope.SQLVars...).Scan(primaryField.Field.Addr().Interface()); scope.Err(err) == nil {

对于连接查询，需要用到预加载github.com/jinzhu/gorm/callback_query.go

DefaultCallback.Query().Register("gorm:preload", preloadCallback)

for fieldIndex, field := range selectFields {

      if field.DBName == column {
        if field.Field.Kind() == reflect.Ptr {
          values[index] = field.Field.Addr().Interface()
          
            func preloadCallback(scope *Scope) {
  if _, skip := scope.InstanceGet("gorm:skip_query_callback"); skip {
    return
  }


  if ap, ok := scope.Get("gorm:auto_preload"); ok {
    // If gorm:auto_preload IS NOT a bool then auto preload.
    // Else if it IS a bool, use the value
    if apb, ok := ap.(bool); !ok {
      autoPreload(scope)
    } else if apb {
      autoPreload(scope)
    }
  }


  if scope.Search.preload == nil || scope.HasError() {
    return
  }


  var (
    preloadedMap = map[string]bool{}
    fields       = scope.Fields()
  )


  for _, preload := range scope.Search.preload {
    var (
      preloadFields = strings.Split(preload.schema, ".")
      currentScope  = scope
      currentFields = fields
    )


    for idx, preloadField := range preloadFields {
      var currentPreloadConditions []interface{}


      if currentScope == nil {
        continue
      }


      // if not preloaded
      if preloadKey := strings.Join(preloadFields[:idx+1], "."); !preloadedMap[preloadKey] {


        // assign search conditions to last preload
        if idx == len(preloadFields)-1 {
          currentPreloadConditions = preload.conditions
        }


        for _, field := range currentFields {
          if field.Name != preloadField || field.Relationship == nil {
            continue
          }


          switch field.Relationship.Kind {
          case "has_one":
            currentScope.handleHasOnePreload(field, currentPreloadConditions)
          case "has_many":
            currentScope.handleHasManyPreload(field, currentPreloadConditions)
          case "belongs_to":
            currentScope.handleBelongsToPreload(field, currentPreloadConditions)
          case "many_to_many":
            currentScope.handleManyToManyPreload(field, currentPreloadConditions)
          default:
            scope.Err(errors.New("unsupported relation"))
          }


          preloadedMap[preloadKey] = true
          break
        }


        if !preloadedMap[preloadKey] {
          scope.Err(fmt.Errorf("can't preload field %s for %s", preloadField, currentScope.GetModelStruct().ModelType))
          return
        }
      }


      // preload next level
      if idx < len(preloadFields)-1 {
        currentScope = currentScope.getColumnAsScope(preloadField)
        if currentScope != nil {
          currentFields = currentScope.Fields()
        }
      }
    }
  }
}

GORM是一个负重前行的框架：它不仅支持了所有原生SQL的特性，也增加了很多类似Hook的高级特性，导致这个框架非常庞大。如果团队没有历史包袱，更推荐节制地使用GORM特性，适当封装一层；interface{}问题 - GORM中许多函数入参的数据类型都是interface{}，底层又用reflect支持了多种类型，这种实现会导致两个问题：

1，reflect导致的底层的性能不高（这点还能接受）

2，interface{}如果传入了不支持的复杂数据类型时，排查问题麻烦，往往要运行程序时才会报错

3，高频拼接重复SQL - 在一个程序运行过程中，执行的SQL语句都比较固定，而变化的往往是参数；从GORM的实现来看，每次执行都需要重新拼接一次SQL语句，是有不小的优化空间的，比如引入一定的cache。或者使用sqlc：https://github.com/xiazemin/sqlc

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2022-04-09，如有侵权请联系 cloudcommunity@tencent.com 删除

github