Gorm 源碼分析(一) database/sql

時間 2019-12-06

標籤 gorm 源碼分析 database sql 欄目 SQL 简体版

原文原文鏈接

簡介

Gorm是Go語言開發用的比較多的一個ORM。它的功能比較全：mysql

增刪改查
關聯（包含一個，包含多個，屬於，多對多，多種包含）
CallBacks（建立、保存、更新、刪除、查詢找）以前以後均可以有callback函數
預加載
事務
複合主鍵
日誌

database/sql 包

可是這篇文章中並不會直接看Gorm的源碼，咱們會先從database/sql分析。緣由是Gorm也是基於這個包來封裝的一些功能。因此只有先了解了database/sql包才能更加好的理解Gorm源碼。
database/sql 其實也是一個對於mysql驅動的上層封裝。"github.com/go-sql-driver/mysql"就是一個對於mysql的驅動，database/sql 就是在這個基礎上作的基本封裝包含鏈接池的使用git

使用例子

下面這個是最基本的增刪改查操做
操做分下面幾個步驟：github

引入github.com/go-sql-driver/mysql包（包中的init方法會初始化mysql驅動的註冊）
使用sql.Open 初始化一個sql.DB結構
調用Prepare Exec 執行sql語句

==注意:==使用Exec函數無需釋放調用完畢以後會自動釋放，把鏈接放入鏈接池中sql

使用Query 返回的sql.rows 須要手動釋放鏈接 rows.Close()

package main

import (
    "database/sql"
    "fmt"
    _ "github.com/go-sql-driver/mysql"
    "strconv"
)

func main() {
    // 打開鏈接
    db, err := sql.Open("mysql", "root:feg@125800@tcp(47.100.245.167:3306)/artifact?charset=utf8&loc=Asia%2FShanghai&parseTime=True")
    if err != nil {
        fmt.Println("err:", err)
    }
    // 設置最大空閒鏈接數
    db.SetMaxIdleConns(1)
    // 設置最大連接數
    db.SetMaxOpenConns(1)
    query(db, 3)
}

//修改
func update(db *sql.DB, id int, user string) {
    stmt, err := db.Prepare("update user set UserName=? where Id =?")
    if err != nil {
        fmt.Println(err)
    }
    res, err := stmt.Exec(user, id)
    updateId, err := res.LastInsertId()
    fmt.Println(updateId)
}

//刪除
func delete(db *sql.DB, id int) {
    stmt, err := db.Prepare("delete  from user where id = ?")
    if err != nil {
        fmt.Println(err)
    }
    res, err := stmt.Exec(1)
    updateId, err := res.LastInsertId()
    fmt.Println(updateId)
}

//查詢
func query(db *sql.DB, id int) {
    rows, err := db.Query("select * from user where  id = " + strconv.Itoa(id))
    if err != nil {
        fmt.Println(err)
        return
    }

    for rows.Next() {
        var id int
        var user string
        var pwd string
        rows.Scan(&id, &user, &pwd)
        fmt.Println("id:", id, "user:", user, "pwd:", pwd)
    }
    rows.Close()
}

//插入
func insert(db *sql.DB, user, pwd string) {
    stmt, err := db.Prepare("insert into user set UserName=?,Password=?")
    if err != nil {
        fmt.Println(err)
    }
    res, err := stmt.Exec("peter", "panlei")
    id, err := res.LastInsertId()
    fmt.Println(id)
}

鏈接池

由於Gorm的鏈接池就是使用database/sql包中的鏈接池，因此這裏咱們須要學習一下包裏的鏈接池的源碼實現。其實全部鏈接池最重要的就是鏈接池對象、獲取函數、釋放函數下面來看一下database/sql中的鏈接池。數據庫

DB對象

type DB struct {
    //數據庫實現驅動
    driver driver.Driver
    dsn    string
    numClosed uint64
    // 鎖
    mu           sync.Mutex // protects following fields
    // 空閒鏈接
    freeConn     []*driverConn
    //阻塞請求隊列，等鏈接數達到最大限制時，後續請求將插入此隊列等待可用鏈接
    connRequests map[uint64]chan connRequest
    // 記錄下一個key用於connRequests map的key
    nextRequest  uint64 // Next key to use in connRequests.
    numOpen      int    // number of opened and pending open connections
    
    openerCh    chan struct{}
    closed      bool
    dep         map[finalCloser]depSet
    lastPut     map[*driverConn]string 
    // 最大空閒鏈接數
    maxIdle     int                    
    // 最大打開鏈接數
    maxOpen     int  
    // 鏈接最大存活時間
    maxLifetime time.Duration          
    cleanerCh   chan struct{}
}

獲取方法

func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
    db.mu.Lock()
    if db.closed {
        db.mu.Unlock()
        return nil, errDBClosed
    }
    // Check if the context is expired.
    select {
    default:
    case <-ctx.Done():
        db.mu.Unlock()
        return nil, ctx.Err()
    }
    lifetime := db.maxLifetime

    // 查看是否有空閒的鏈接 若是有則直接使用空閒鏈接
    numFree := len(db.freeConn)
    if strategy == cachedOrNewConn && numFree > 0 {
        // 取出數據第一個
        conn := db.freeConn[0]
        // 複製數組，去除第一個鏈接
        copy(db.freeConn, db.freeConn[1:])
        db.freeConn = db.freeConn[:numFree-1]
        conn.inUse = true
        db.mu.Unlock()
        if conn.expired(lifetime) {
            conn.Close()
            return nil, driver.ErrBadConn
        }
        return conn, nil
    }

    // 判斷是否超出最大鏈接數 
    if db.maxOpen > 0 && db.numOpen >= db.maxOpen {
        // 建立一個chan 
        req := make(chan connRequest, 1)
        // 獲取下一個request 做爲map 中的key
        reqKey := db.nextRequestKeyLocked()
        db.connRequests[reqKey] = req
        db.mu.Unlock()

        // Timeout the connection request with the context.
        select {
        case <-ctx.Done():
            // Remove the connection request and ensure no value has been sent
            // on it after removing.
            db.mu.Lock()
            delete(db.connRequests, reqKey)
            db.mu.Unlock()
            select {
            default:
            case ret, ok := <-req:
                if ok {
                    db.putConn(ret.conn, ret.err)
                }
            }
            return nil, ctx.Err()
        // 若是沒有取消則從req chan中獲取數據 阻塞主一直等待有conn數據傳入
        case ret, ok := <-req:
            if !ok {
                return nil, errDBClosed
            }
            // 判斷超時 
            if ret.err == nil && ret.conn.expired(lifetime) {
                ret.conn.Close()
                return nil, driver.ErrBadConn
            }
            return ret.conn, ret.err
        }
    }
    
    db.numOpen++ // optimistically
    db.mu.Unlock()
    // 調用driver的Open方法創建鏈接
    ci, err := db.driver.Open(db.dsn)
    if err != nil {
        db.mu.Lock()
        db.numOpen-- // correct for earlier optimism
        db.maybeOpenNewConnections()
        db.mu.Unlock()
        return nil, err
    }
    db.mu.Lock()
    dc := &driverConn{
        db:        db,
        createdAt: nowFunc(),
        ci:        ci,
        inUse:     true,
    }
    db.addDepLocked(dc, dc)
    db.mu.Unlock()
    return dc, nil
}

釋放鏈接方法

// 釋放鏈接
func (db *DB) putConn(dc *driverConn, err error) {
    db.mu.Lock()
    if !dc.inUse {
        if debugGetPut {
            fmt.Printf("putConn(%v) DUPLICATE was: %s\n\nPREVIOUS was: %s", dc, stack(), db.lastPut[dc])
        }
        panic("sql: connection returned that was never out")
    }
    if debugGetPut {
        db.lastPut[dc] = stack()
    }
    // 設置已經在使用中
    dc.inUse = false

    for _, fn := range dc.onPut {
        fn()
    }
    dc.onPut = nil
    // 判斷鏈接是否有錯誤 
    if err == driver.ErrBadConn {
        db.maybeOpenNewConnections()
        db.mu.Unlock()
        dc.Close()
        return
    }
    if putConnHook != nil {
        putConnHook(db, dc)
    }
    // 調用方法 釋放鏈接
    added := db.putConnDBLocked(dc, nil)
    db.mu.Unlock()
    // 判斷若是沒有加到了空閒列表中 dc關閉
    if !added {
        dc.Close()
    }
}

func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
    if db.closed {
        return false
    }
    if db.maxOpen > 0 && db.numOpen > db.maxOpen {
        return false
    }
    // 若是等待chan列表大於0 
    if c := len(db.connRequests); c > 0 {
        var req chan connRequest
        var reqKey uint64
        // 獲取map 中chan和key
        for reqKey, req = range db.connRequests {
            break
        }
        // 從列表中刪除chan 
        delete(db.connRequests, reqKey) // Remove from pending requests.
        if err == nil {
            dc.inUse = true
        }
        // 把鏈接傳入chan中 讓以前獲取鏈接被阻塞的獲取函數繼續
        req <- connRequest{
            conn: dc,
            err:  err,
        }
        return true
    } else if err == nil && !db.closed && db.maxIdleConnsLocked() > len(db.freeConn) {
        // 若是沒有等待列表，則把鏈接放到空閒列表中
        db.freeConn = append(db.freeConn, dc)
        db.startCleanerLocked()
        return true
    }
    return false
}

鏈接池的實現有不少方法，在database/sql包中使用的是chan阻塞使用map記錄等待列表，等到有鏈接釋放的時候再把鏈接傳入等待列表中的chan 不在阻塞返回鏈接。
以前咱們看到的Redigo是使用一個chan 來阻塞，而後釋放的時候放入空閒列表，在往這一個chan中傳入struct{}{}，讓程序繼續獲取的時候再從空閒列表中獲取。而且使用的是鏈表的結構來存儲空閒列表。數組