golang sql語(yǔ)句超時(shí)控制方案及原理

更新時(shí)間：2023年12月18日 11:25:43 作者：hayson

一般應(yīng)用程序在執(zhí)行一條sql語(yǔ)句時(shí),都會(huì)給這條sql設(shè)置一個(gè)超時(shí)時(shí)間,本文主要介紹了golang sql語(yǔ)句超時(shí)控制方案及原理,具有一定的參考價(jià)值,感興趣的可以了解一下

一般應(yīng)用程序在執(zhí)行一條sql語(yǔ)句時(shí)，都會(huì)給這條sql設(shè)置一個(gè)超時(shí)時(shí)間，如果到超時(shí)時(shí)間還未執(zhí)行完，則直接終止sql，釋放資源，返回錯(cuò)誤。這里主要討論一下在golang+mysql的場(chǎng)景下，對(duì)sql語(yǔ)句進(jìn)行超時(shí)控制的具體做法、實(shí)現(xiàn)原理以及對(duì)連接池連接數(shù)產(chǎn)生的影響。

基于context實(shí)現(xiàn)sql語(yǔ)句的超時(shí)控制：

使用context進(jìn)行超時(shí)控制是golang的標(biāo)準(zhǔn)做法，可以說(shuō)當(dāng)一個(gè)函數(shù)第一個(gè)參數(shù)是ctx context.Context時(shí)，這個(gè)函數(shù)就應(yīng)該做出承諾，在收到ctx的取消信號(hào)時(shí)應(yīng)該提前終止該函數(shù)的執(zhí)行，并釋放資源。目前后端應(yīng)用程序操作數(shù)據(jù)庫(kù)時(shí)比較常用的做法是使用gorm框架，這個(gè)框架主要是起到sql拼接和屏蔽底層數(shù)據(jù)庫(kù)差異的作用，本身并沒(méi)有提供連接池以及mysql的client端驅(qū)動(dòng)程序，連接池默認(rèn)使用的是database/sql標(biāo)準(zhǔn)庫(kù)提供的連接池，驅(qū)動(dòng)程序使用的是go-sql-driver/mysql。故要分析如何基于context進(jìn)行超時(shí)控制，需要從這三層進(jìn)行分析。

對(duì)于gorm，想要對(duì)一個(gè)sql進(jìn)行超時(shí)控制，可以直接使用WithContext()方法，具體如下：

func main() {
  ctx := context.TODO()
  ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  defer cancel()
  err := db.WithContext(ctx).Exec("select sleep(10)").Error
  if err != nil {
    log.Fatal(err)
  }
}
?
// output
// [3001.379ms] [rows:0] select sleep(10)
// 2023/12/17 13:31:54 context deadline exceeded

這里將ctx的超時(shí)時(shí)間設(shè)置為3s，同時(shí)sql語(yǔ)句為sleep 10s，最終在執(zhí)行時(shí)間到3s時(shí)，返回了context deadline exceeded錯(cuò)誤。

gorm調(diào)用WithContext之后，最終會(huì)將這個(gè)ctx給到database/sql連接池的ExecContext函數(shù)之中，

func (db *DB) ExecContext(ctx context.Context, query string, args ...any) (Result, error) {
  var res Result
  var err error
?
  err = db.retry(func(strategy connReuseStrategy) error {
    res, err = db.exec(ctx, query, args, strategy)
    return err
  })
?
  return res, err
}

該函數(shù)會(huì)從連接池中取出一個(gè)連接，然后由數(shù)據(jù)庫(kù)驅(qū)動(dòng)層實(shí)際執(zhí)行sql，

func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
    dargs, err := namedValueToValue(args)
    if err != nil {
       return nil, err
    }
    // 這里是核心代碼，將ctx放入監(jiān)聽(tīng)隊(duì)列
    if err := mc.watchCancel(ctx); err != nil {
       return nil, err
    }
    defer mc.finish()
?
    return mc.Exec(query, dargs)
}

go-sql-drive/mysql在實(shí)際和mysql server端通信之前，會(huì)調(diào)用watchCancel，監(jiān)聽(tīng)當(dāng)前ctx的取消信號(hào)，保證sql執(zhí)行過(guò)程中，能夠立刻收到取消信號(hào)，并做出sql取消的操作。watchCancel函數(shù)具體實(shí)現(xiàn)如下：

func (mc *mysqlConn) watchCancel(ctx context.Context) error {
    if mc.watching {
       // Reach here if canceled,
       // so the connection is already invalid
       mc.cleanup()
       return nil
    }
    // When ctx is already cancelled, don't watch it.
    if err := ctx.Err(); err != nil {
       return err
    }
    // When ctx is not cancellable, don't watch it.
    if ctx.Done() == nil {
       return nil
    }
    // When watcher is not alive, can't watch it.
    if mc.watcher == nil {
       return nil
    }
?
    // 在正常情況下會(huì)走到這里，將ctx放入到一個(gè)watcher管道
    mc.watching = true
    mc.watcher <- ctx
    return nil
}

可以看到核心代碼是將這個(gè)ctx放入到一個(gè)管道，那么必定有一段程序是監(jiān)聽(tīng)這個(gè)管道的，實(shí)際上是如下代碼：

func (mc *mysqlConn) startWatcher() {
    watcher := make(chan context.Context, 1)
    mc.watcher = watcher
    finished := make(chan struct{})
    mc.finished = finished
    go func() {
       for {
          var ctx context.Context
          select {
          case ctx = <-watcher:
          case <-mc.closech:
             return
          }
?
          select {
          // 在這里監(jiān)聽(tīng)了ctx取消信號(hào)，并實(shí)際執(zhí)行cancel操作
          case <-ctx.Done():
             mc.cancel(ctx.Err())
          case <-finished:
          case <-mc.closech:
             return
          }
       }
    }()
}

這個(gè)startWatcher函數(shù)內(nèi)部會(huì)單獨(dú)啟動(dòng)一個(gè)協(xié)程，監(jiān)聽(tīng)本連接的watcher管道，針對(duì)于每一個(gè)從管道中取出的ctx，監(jiān)聽(tīng)其取消信號(hào)是否結(jié)束，同一個(gè)連接上的sql語(yǔ)句肯定是依次執(zhí)行的，這樣依次監(jiān)聽(tīng)每一個(gè)ctx是不會(huì)有什么問(wèn)題的，而這個(gè)startWatcher會(huì)在連接創(chuàng)建的時(shí)候調(diào)用，保證后續(xù)這個(gè)連接上的每個(gè)語(yǔ)句添加的ctx都會(huì)被監(jiān)聽(tīng)。

如果真的監(jiān)聽(tīng)到取消信號(hào)，就會(huì)調(diào)用cancel函數(shù)進(jìn)行取消，

// finish is called when the query has canceled.
func (mc *mysqlConn) cancel(err error) {
    mc.canceled.Set(err)
    mc.cleanup()
}
?
func (mc *mysqlConn) cleanup() {
  if !mc.closed.TrySet(true) {
    return
  }
?
  // Makes cleanup idempotent
  close(mc.closech)
  if mc.netConn == nil {
    return
  }
  // 核心代碼如下，關(guān)閉了通信所使用的TCP連接
  if err := mc.netConn.Close(); err != nil {
    errLog.Print(err)
  }
}

最終在收到取消信號(hào)時(shí)，會(huì)關(guān)閉和mysql server進(jìn)行通信的TCP連接。

基于DSN中的readTimeout和writeTimeout實(shí)現(xiàn)sql語(yǔ)句的超時(shí)控制：

還有一種方法是在打開(kāi)一個(gè)db對(duì)象的dsn中指定，具體做法如下:

func init() {
    dsn := "root:12345678@tcp(localhost:3306)/test?charset=utf8mb4&parseTime=True&loc=Local&timeout=1500ms&readTimeout=3s&writeTimeout=3s"
    var err error
    db, err = gorm.Open(mysql.Open(dsn), &gorm.Config{})
    if err != nil {
       log.Fatalln(err)
    }
    db.Logger.LogMode(logger.Info)
}
?
func main() {
  ctx := context.TODO()
  err := db.WithContext(ctx).Exec("select sleep(10)").Error
  if err != nil {
    log.Fatal(err)
  }
}
?
// output
// [3002.597ms] [rows:0] select sleep(10)
// 2023/12/17 14:21:11 invalid connection

在dsn中指定readTimeout=3s&writeTimeout=3s，同時(shí)執(zhí)行一個(gè)sleep(10)，同樣可以在第三秒時(shí)報(bào)錯(cuò)，但報(bào)錯(cuò)會(huì)有一點(diǎn)點(diǎn)奇怪，invalid connection，看起來(lái)好像和超時(shí)沒(méi)有啥關(guān)系，這是因?yàn)檫@兩個(gè)超時(shí)時(shí)間的含義其實(shí)是針對(duì)于mysql底層使用的TCP連接而言的，即readTimeout是從TCP連接中讀取一個(gè)數(shù)據(jù)包的超時(shí)時(shí)間，writeTimeout是向一個(gè)TCP連接寫入一個(gè)數(shù)據(jù)包的超時(shí)時(shí)間，并且這個(gè)超時(shí)是基于連接的deadline實(shí)現(xiàn)的，所以一旦超時(shí)就會(huì)認(rèn)為這個(gè)連接是異常的，最終返回這樣一個(gè)連接異常的報(bào)錯(cuò)。

具體的實(shí)現(xiàn)原理仍然是在go-sql-driver/mysql中，在創(chuàng)建連接時(shí)，會(huì)處理這兩個(gè)timeout，

...
?
// 這就是上面說(shuō)的啟動(dòng)監(jiān)聽(tīng)ctx的邏輯
mc.startWatcher()
if err := mc.watchCancel(ctx); err != nil {
    mc.cleanup()
    return nil, err
}
defer mc.finish()
?
mc.buf = newBuffer(mc.netConn)
?
// 在解析了dsn之后，就會(huì)將兩個(gè)timeout賦值給核心連接對(duì)象的兩個(gè)屬性
mc.buf.timeout = mc.cfg.ReadTimeout
mc.writeTimeout = mc.cfg.WriteTimeout
?
...

在實(shí)際和mysql server進(jìn)行通信時(shí)，會(huì)用到這兩個(gè)屬性，

func (b *buffer) readNext(need int) ([]byte, error) {
    if b.length < need {
       // refill
       if err := b.fill(need); err != nil {
          return nil, err
       }
    }
?
    offset := b.idx
    b.idx += need
    b.length -= need
    return b.buf[offset:b.idx], nil
}
?
// fill reads into the buffer until at least _need_ bytes are in it
func (b *buffer) fill(need int) error {
  
  ...go
  
  for {
    // 若timeout>0，則基于這個(gè)超時(shí)時(shí)間，給連接設(shè)置一個(gè)新的deadline
    if b.timeout > 0 {
      if err := b.nc.SetReadDeadline(time.Now().Add(b.timeout)); err != nil {
        return err
      }
    }
?
    nn, err := b.nc.Read(b.buf[n:])
    n += nn
?
    switch err {
    case nil:
      if n < need {
        continue
      }
      b.length = n
      return nil
?
    case io.EOF:
      if n >= need {
        b.length = n
        return nil
      }
      return io.ErrUnexpectedEOF
?
    default:
      return err
    }
  }
}

在每次需要從mysql server獲取數(shù)據(jù)的時(shí)候，都會(huì)給這次讀操作設(shè)置一個(gè)deadline，具體的時(shí)間就是當(dāng)前時(shí)間+timeout值，這樣每次從server端讀取數(shù)據(jù)的時(shí)候，一旦超出這個(gè)時(shí)間，就會(huì)報(bào)一個(gè)io timeout錯(cuò)誤，而上游再收到這個(gè)錯(cuò)誤之后，則會(huì)進(jìn)行如下處理：

data, err = mc.buf.readNext(pktLen)
if err != nil {
    if cerr := mc.canceled.Value(); cerr != nil {
       return nil, cerr
    }
    errLog.Print(err)
    // 關(guān)閉當(dāng)前連接
    mc.Close()
    // 返回invalid connection錯(cuò)誤
    return nil, ErrInvalidConn
}

首先將這條連接關(guān)閉，之后返回了invalid connection錯(cuò)誤，這也就是為什么上面例子中超時(shí)的報(bào)錯(cuò)是invalid connection。核心需要關(guān)注的還是Close方法，這里是超時(shí)后續(xù)的處理：

func (mc *mysqlConn) Close() (err error) {
    // Makes Close idempotent
    if !mc.closed.IsSet() {
      // 向mysql server發(fā)送quit命令，表明自己要退出了
       err = mc.writeCommandPacket(comQuit)
    }
?
    // 調(diào)用cleanup，和上面監(jiān)聽(tīng)ctx取消信號(hào)后的操作是一致的
    mc.cleanup()
?
    return
}

首先發(fā)送一個(gè)quit命令，告知自己需要退出，之后也使用cleanup方法，關(guān)閉tcp連接?？梢钥吹竭@里的超時(shí)控制邏輯和基于ctx的對(duì)比，基本是一致的，就是目前這種方案還給server端發(fā)送了一個(gè)quit指令，從外部使用上看似乎加不加這個(gè)指令效果都是一樣的，只要連接關(guān)閉了mysql server端就可以回收自己的資源（不一定能立刻回收，但最終會(huì)回收）。我查找了一些資料和mysql的官方文檔，并沒(méi)有找到如果不發(fā)送quit指令，直接關(guān)閉tcp連接會(huì)有什么影響，我也沒(méi)有研究過(guò)mysql的源碼，如果有人知道的話，還請(qǐng)不吝賜教。但我想應(yīng)該沒(méi)有太大問(wèn)題，要不然基于ctx的超時(shí)控制早就出問(wèn)題了。

sql語(yǔ)句超時(shí)時(shí)連接池如何處理：

以上兩種sql超時(shí)的方案我個(gè)人覺(jué)得都沒(méi)有什么問(wèn)題，底層最終面對(duì)超時(shí)時(shí)所做的操作也基本一致（關(guān)閉TCP連接），我個(gè)人更喜歡基于ctx的方案，畢竟ctx設(shè)計(jì)之初就是用來(lái)做這件事的，也可以和其他場(chǎng)景下的超時(shí)控制保持一致，報(bào)錯(cuò)信息也更友好一些。接下來(lái)需要考慮的就是一旦底層出現(xiàn)報(bào)錯(cuò)，連接被關(guān)閉，上層的連接池是如何處理的，

func (db *DB) execDC(ctx context.Context, dc *driverConn, release func(error), query string, args []any) (res Result, err error) {
    defer func() {
       // 核心代碼在這里，執(zhí)行完sql之后需要釋放當(dāng)前連接，釋放時(shí)會(huì)基于err是否為nil做出處理
       release(err)
    }()
    execerCtx, ok := dc.ci.(driver.ExecerContext)
    var execer driver.Execer
    if !ok {
       execer, ok = dc.ci.(driver.Execer)
    }
    if ok {
       var nvdargs []driver.NamedValue
       var resi driver.Result
       withLock(dc, func() {
          nvdargs, err = driverArgsConnLocked(dc.ci, nil, args)
          if err != nil {
             return
          }
          // 驅(qū)動(dòng)層實(shí)際進(jìn)行查詢
          resi, err = ctxDriverExec(ctx, execerCtx, execer, query, nvdargs)
       })
       if err != driver.ErrSkip {
          if err != nil {
             return nil, err
          }
          return driverResult{dc, resi}, nil
       }
    }
    ...
}

exec執(zhí)行完之后，會(huì)釋放該連接，將其放回連接池，以供其他查詢使用，放回連接池時(shí)會(huì)依據(jù)本條sql是否有錯(cuò)誤進(jìn)行處理，release函數(shù)時(shí)注入進(jìn)來(lái)的，實(shí)際上是releaseConn函數(shù)，該函數(shù)內(nèi)部調(diào)用了putConn函數(shù)，

func (db *DB) putConn(dc *driverConn, err error, resetSession bool) {
    if !errors.Is(err, driver.ErrBadConn) {
       // 這里判斷了一下連接是不是已經(jīng)不可用了，若已經(jīng)不可用則將err賦值為ErrBadConn
       if !dc.validateConnection(resetSession) {
          err = driver.ErrBadConn
       }
    }
    db.mu.Lock()
    if !dc.inUse {
       db.mu.Unlock()
       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")
    }
    // 若連接已到最大生存時(shí)間，也要標(biāo)記連接已經(jīng)不可用
    if !errors.Is(err, driver.ErrBadConn) && dc.expired(db.maxLifetime) {
       db.maxLifetimeClosed++
       err = driver.ErrBadConn
    }
    if debugGetPut {
       db.lastPut[dc] = stack()
    }
    dc.inUse = false
    dc.returnedAt = nowFunc()
?
    for _, fn := range dc.onPut {
       fn()
    }
    dc.onPut = nil
    // 若連接不可用，進(jìn)行如下處理
    if errors.Is(err, driver.ErrBadConn) {
       // 有一個(gè)連接被關(guān)閉，考慮打開(kāi)一個(gè)新的連接
       db.maybeOpenNewConnections()
       db.mu.Unlock()
       // 關(guān)閉該連接
       dc.Close()
       return
    }
    if putConnHook != nil {
       putConnHook(db, dc)
    }
    // sql執(zhí)行正常，或者有一些錯(cuò)誤但連接是正常的，會(huì)正常的歸還連接
    added := db.putConnDBLocked(dc, nil)
    db.mu.Unlock()
?
    if !added {
       dc.Close()
       return
    }
}

整體而言，在sql執(zhí)行出現(xiàn)異常時(shí)，會(huì)判斷一下連接是否可用，這個(gè)判斷也是于驅(qū)動(dòng)層完成，驅(qū)動(dòng)層實(shí)現(xiàn)了如下方法：

// IsValid implements driver.Validator interface
// (From Go 1.15)
func (mc *mysqlConn) IsValid() bool {
    return !mc.closed.IsSet()
}

用于告知連接池這個(gè)連接是否還正常，而在sql執(zhí)行超時(shí)最后調(diào)用的cleanup方法里，首先就是標(biāo)記這個(gè)連接已經(jīng)不可用了

func (mc *mysqlConn) cleanup() {
  // 標(biāo)記連接不可用
  if !mc.closed.TrySet(true) {
    return
  }
}

在判斷連接異常，或者超出最大生存時(shí)間之后，就是調(diào)用連接池的Close方法，注意是連接池的Close，不是驅(qū)動(dòng)層的Close，這個(gè)Close最終會(huì)調(diào)用到finalClose。

func (dc *driverConn) finalClose() error {
    var err error
    var openStmt []*driverStmt
    withLock(dc, func() {
       openStmt = make([]*driverStmt, 0, len(dc.openStmt))
       for ds := range dc.openStmt {
          openStmt = append(openStmt, ds)
       }
       dc.openStmt = nil
    })
    for _, ds := range openStmt {
       ds.Close()
    }
    withLock(dc, func() {
       // 這里調(diào)用驅(qū)動(dòng)層進(jìn)行連接的關(guān)閉
       dc.finalClosed = true
       err = dc.ci.Close()
       dc.ci = nil
    })
?
    dc.db.mu.Lock()
    // 當(dāng)前打開(kāi)連接數(shù)減一
    dc.db.numOpen--
    dc.db.maybeOpenNewConnections()
    dc.db.mu.Unlock()
?
    dc.db.numClosed.Add(1)
    return err
}

這個(gè)方法主要是做兩件事，一是在驅(qū)動(dòng)層實(shí)際關(guān)閉連接，這主要是針對(duì)達(dá)到最大生存時(shí)間的連接，對(duì)于sql執(zhí)行超時(shí)這種本身就已經(jīng)關(guān)閉了的連接是不會(huì)再關(guān)閉一次的，TrySet會(huì)執(zhí)行不成功，后面TCP鏈接關(guān)閉的操作是不會(huì)繼續(xù)執(zhí)行的。關(guān)閉連接之后，讓當(dāng)前打開(kāi)的連接數(shù)減一，從而保證可以正常打開(kāi)新的連接。

有可能帶來(lái)的問(wèn)題：

綜上所述，這兩種超時(shí)控制的方法實(shí)現(xiàn)原理雖然有所區(qū)別，但在發(fā)現(xiàn)超時(shí)后做的事情是一致的，都是關(guān)閉該連接，并且讓連接池打開(kāi)連接數(shù)量減一，這其實(shí)存在著一個(gè)問(wèn)題，因?yàn)閏lient端雖然正常調(diào)了Close，認(rèn)為連接已經(jīng)關(guān)閉了，但其實(shí)mysql server端在非sleep狀態(tài)下是感知不到連接關(guān)閉的消息的，一種具體的情況就是比如mysql的某個(gè)連接正在執(zhí)行一個(gè)耗時(shí)的查詢，但是這時(shí)到了超時(shí)時(shí)間，client主動(dòng)關(guān)閉了連接，但是mysql server端是不會(huì)立刻終止查詢并關(guān)閉連接的，show processlist時(shí)，仍然能看到連接中的sql還在正常執(zhí)行。其實(shí)觀察TCP連接的狀態(tài)也能看到這一現(xiàn)象，在雙方正常通信時(shí)，狀態(tài)為

tcp6 0 0 ::1.3306 ::1.61351 ESTABLISHED
tcp6 0 0 ::1.61351 ::1.3306 ESTABLISHED

在client端主動(dòng)調(diào)用Close之后，server端由于要執(zhí)行當(dāng)前sql會(huì)一直保持在CLOSE_WAIT狀態(tài)，client端進(jìn)入FIN_WAIT_2狀態(tài)，直到server端在sql執(zhí)行完成后才會(huì)進(jìn)行后續(xù)的揮手過(guò)程，才能真正關(guān)閉連接。

tcp6 5 0 ::1.3306 ::1.61351 CLOSE_WAIT
tcp6 0 0 ::1.61351 ::1.3306 FIN_WAIT_2

問(wèn)題就在于server端連接還未關(guān)閉，但連接池那邊連接數(shù)已經(jīng)減一了，后續(xù)可以創(chuàng)建新的連接了，這就導(dǎo)致mysql server端的連接數(shù)是會(huì)高于連接池的最大連接數(shù)的，如果超時(shí)的sql很多，很有可能導(dǎo)致連接數(shù)超出連接池最大連接數(shù)限制達(dá)到mysql server端的最大連接數(shù)，后續(xù)新的連接將無(wú)法建立，直接返回too many connectios錯(cuò)誤，如果這些連接執(zhí)行的sql又真的很慢，或者發(fā)生死鎖，可能會(huì)出現(xiàn)mysql較長(zhǎng)時(shí)間直接拒絕服務(wù)的情況。這表明在生產(chǎn)環(huán)境下盡量不要將mysql的max connectios參數(shù)設(shè)置的和數(shù)據(jù)庫(kù)最大連接數(shù)比較接近，還是要留出一定的余量，避免在出現(xiàn)很多sql超時(shí)時(shí)這部分泄露的連接直接將mysql連接數(shù)打滿，導(dǎo)致數(shù)據(jù)庫(kù)出現(xiàn)不可用。

到此這篇關(guān)于golang sql語(yǔ)句超時(shí)控制方案及原理的文章就介紹到這了,更多相關(guān)golang sql超時(shí)控制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: