本文作者 Xargin，個人博客：https://xargin.com/。

http 標準庫

服務端

請求處理

package main

import (
 "io"
 "log"
 "net/http"
)

func sayhello(wr http.ResponseWriter, r *http.Request) {
 wr.Header()["Content-Type"] = []string{"Application/json"}
 io.WriteString(wr, "hello")
}

func main() {
 http.HandleFunc("/", sayhello)
 http.ListenAndServe(":9090", nil)
}

1-1

每一個請求啟動一個 goroutine，讀取完畢之后，調用用戶傳入的 handler(沒有的話就用默認的)，在同一連接進行 response 響應。整體上是個 request/response loop 模型。

客戶端

連接池

type Transport struct {
 idleMu       sync.Mutex
 closeIdle    bool                                // user has requested to close all idle conns
 idleConn     map[connectMethodKey][]*persistConn // most recently used at end
 idleConnWait map[connectMethodKey]wantConnQueue  // waiting getConns
 idleLRU      connLRU

 connsPerHostMu   sync.Mutex
 connsPerHost     map[connectMethodKey]int
 connsPerHostWait map[connectMethodKey]wantConnQueue // waiting getConns

    // MaxIdleConns controls the maximum number of idle (keep-alive)
 // connections across all hosts. Zero means no limit.
 MaxIdleConns int

 // MaxIdleConnsPerHost, if non-zero, controls the maximum idle
 // (keep-alive) connections to keep per-host. If zero,
 // DefaultMaxIdleConnsPerHost is used.
 MaxIdleConnsPerHost int

 // MaxConnsPerHost optionally limits the total number of
 // connections per host, including connections in the dialing,
 // active, and idle states. On limit violation, dials will block.
 //
 // Zero means no limit.
 MaxConnsPerHost int

 // IdleConnTimeout is the maximum amount of time an idle
 // (keep-alive) connection will remain idle before closing
 // itself.
 // Zero means no limit.
 IdleConnTimeout time.Duration
}

transport 和 client 是一一對應，每個 tranport 內有自己的 connpool， idleConn 的結構是：map[connectMethodKey][]*persistConn，這個 map 的 key 是個數據結構：

// connectMethodKey is the map key version of connectMethod, with a
// stringified proxy URL (or the empty string) instead of a pointer to
// a URL.
type connectMethodKey struct {
 proxy, scheme, addr string
 onlyH1              bool
}

proxy 地址 + 協議 + 地址，以及是否只支持 http1，構成該 map 的 key，proxy 地址是完整的 proxy 地址，比如 export HTTP_PROXY=localhost:1081，則該地址為用戶提供的字符串。scheme 一般是 http:// 或 https:// 之類的字符串，addr 包含完整的域名(或 IP)和端口。

getConn：

2

在 http2 中，同一個連接可以被重復使用，所以 http2 的邏輯里，該連接被返回后仍然保持在連接池里。是否可以重復使用由 pconn.alt 來決定。

tryPutIdleConn

3

如果有正在等待連接的 goroutine，那么就把這條連接 deliver 給相應的 goroutine，這會觸發相應的 ready 操作，使阻塞中的 goroutine 被喚醒繼續處理請求。

否則將連接放回到 Transport 的 idleConn 和 idleLRU 中。

readloop 和 writeloop

func (t *Transport) dialConn(ctx context.Context, cm connectMethod) (pconn *persistConn, err error) {
 go pconn.readLoop()
 go pconn.writeLoop()
 return pconn, nil
}

所以每個 conn 都會有相應的 readloop 和 writeloop，因此每個連接至少有兩個 goroutine。

用戶協程在使用 http.Client 發送請求時，一路到 http.Transport.roundTrip -> http.persistConn.roundTrip：

 pc.writech <- writeRequest{req, writeErrCh, continueCh}

 resc := make(chan responseAndError)
 pc.reqch <- requestAndChan{
  req:        req.Request,
  ch:         resc,
  addedGzip:  requestedGzip,
  continueCh: continueCh,
  callerGone: gone,
 }

在該函數中，將 request 和接收請求的 ch 傳入到 reqch，把 writeRequest 寫入到 writech。

• writeloop 從 writech 中收到了寫請求，會把內容寫入到 conn 上，這個請求也就發給 server 端了
• readloop 收到 requestAndChan 結果，上面 writeloop 相當于已經把請求數據發送到 server 端，readloop 這時候可以從 conn 上讀出 server 發回的 response 數據，所以 readloop 主要做的就是 ReadResponse，然后把 response 的內容寫入到 requestAndChan.ch 中。
• 主協程只要監聽 requestAndChan.ch 來接收相應的 response 即可(用 select 同時監聽 err、連接關閉等 chan)。

這里 http 標準庫的做法要參考一下，把接收數據和相應的錯誤處理代碼可以都集中在一起：

 for {
  testHookWaitResLoop()
  select {
  case err := <-writeErrCh: // 往 server 端寫數據異常
   if debugRoundTrip {
    req.logf("writeErrCh resv: %T/%#v", err, err)
   }
   if err != nil {
    pc.close(fmt.Errorf("write error: %v", err))
    return nil, pc.mapRoundTripError(req, startBytesWritten, err)
   }
   if d := pc.t.ResponseHeaderTimeout; d > 0 {
    if debugRoundTrip {
     req.logf("starting timer for %v", d)
    }
    timer := time.NewTimer(d)
    defer timer.Stop() // prevent leaks
    respHeaderTimer = timer.C
   }
  case <-pc.closech: // 連接關閉異常
   if debugRoundTrip {
    req.logf("closech recv: %T %#v", pc.closed, pc.closed)
   }
   return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
  case <-respHeaderTimer: // 讀請求頭超時
   if debugRoundTrip {
    req.logf("timeout waiting for response headers.")
   }
   pc.close(errTimeout)
   return nil, errTimeout
  case re := <-resc: // 正常地從 response 的 channel 里讀到了響應數據
   if (re.res == nil) == (re.err == nil) {
    panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
   }
   if debugRoundTrip {
    req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
   }
   if re.err != nil {
    return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
   }
   return re.res, nil
  case <-cancelChan: // 用戶側通過 context 取消了流程
   pc.t.CancelRequest(req.Request)
   cancelChan = nil
  case <-ctxDoneChan: // 這個應該意思差不多
   pc.t.cancelRequest(req.Request, req.Context().Err())
   cancelChan = nil
   ctxDoneChan = nil
  }
 }

http2

https://tools.ietf.org/html/rfc7540 https://github.com/bagder/http2-explained

4

http2 協議通過 frame 中的 stream id 對請求和響應進行關聯。

http2 可以不等待上一個請求響應后再發下一個請求，因此同一個連接上可以實現 multiplexing。標準庫中對于 http2 連接的處理復用了 http1 的連接池邏輯，只不過從連接池中取連接時，并沒有真的從連接池里把這個連接拿走。獲取到的連接依然保留在 connpool 中。

除此之外，h2 的 connpool 和 h1 的沒什么區別。

從 idleConn 數組中獲取 idle 連接時：

func (t *Transport) queueForIdleConn(w *wantConn) (delivered bool) {
   if delivered {
    if pconn.alt != nil {
     // HTTP/2: multiple clients can share pconn.
     // Leave it in the list.
    } else {
     // HTTP/1: only one client can use pconn.
     // Remove it from the list.
     t.idleLRU.remove(pconn)
     list = list[:len(list)-1]
    }
   }

把使用完的連接放回連接池時：

 // HTTP/2 (pconn.alt != nil) connections do not come out of the idle list,
 // because multiple goroutines can use them simultaneously.
 // If this is an HTTP/2 connection being “returned,” we're done.
 if pconn.alt != nil && t.idleLRU.m[pconn] != nil {
  return nil
 }

 if pconn.alt == nil {
   // HTTP/1.
   // Loop over the waiting list until we find a w that isn't done already, and hand it pconn.
   for q.len() > 0 {
    w := q.popFront()
    if w.tryDeliver(pconn, nil) {
     done = true
     break
    }
   }
  } else {
   // HTTP/2.
   // Can hand the same pconn to everyone in the waiting list,
   // and we still won't be done: we want to put it in the idle
   // list unconditionally, for any future clients too.
   for q.len() > 0 {
    w := q.popFront()
    w.tryDeliver(pconn, nil)
   }
  }

• 如果 LRU 列表非空，說明當前沒有等待的 goroutine，而在獲取 http2 連接時，并沒有把連接從連接池中真地拿走，所以直接返回就行了。
• 如果 LRU 列表為空，這條可能是新建的連接，需要把 waitqueue 彈到空，并把當前這條連接放進連接池。

fasthttp

服務端

請求處理

5

fasthttp 的 server 端使用 worker pool 來進行 goroutine 復用，不會頻繁創建新的 g。

workerPool.workerFunc 就是每個 worker 的主循環：

func (wp *workerPool) workerFunc(ch *workerChan) {
 var c net.Conn

 for c = range ch.ch {
  if c == nil {
   break
  }

  wp.WorkerFunc(c)
 }

 wp.lock.Lock()
 wp.workersCount--
 wp.lock.Unlock()
}

每次 serve 新的 conn 時：

1. 從 workerpool 中獲取一個 worker，沒有就新建，啟動 workerFunc 主循環，監聽 worker channel。
2. 把當前 serve 的新連接發送到 worker channel
3. workerFunc 獲取到新 conn，即開始請求處理流程。執行 fasthttp.Server.serveConn

客戶端

連接池

type HostClient struct {
 // Maximum number of connections which may be established to all hosts
 // listed in Addr.
 //
 // You can change this value while the HostClient is being used
 // using HostClient.SetMaxConns(value)
 //
 // DefaultMaxConnsPerHost is used if not set.
 MaxConns int

 // Keep-alive connections are closed after this duration.
 //
 // By default connection duration is unlimited.
 MaxConnDuration time.Duration

 // Idle keep-alive connections are closed after this duration.
 //
 // By default idle connections are closed
 // after DefaultMaxIdleConnDuration.
 MaxIdleConnDuration time.Duration

 // Maximum number of attempts for idempotent calls
 //
 // DefaultMaxIdemponentCallAttempts is used if not set.
 MaxIdemponentCallAttempts int

    
    // Maximum duration for waiting for a free connection.
 //
 // By default will not waiting, return ErrNoFreeConns immediately
 MaxConnWaitTimeout time.Duration

 clientName  atomic.Value
 lastUseTime uint32

 connsLock  sync.Mutex
 connsCount int
 conns      []*clientConn
 connsWait  *wantConnQueue
}

acquireConn

6

流程比較簡單，如果當前 client.conns 數組 > 0，說明有空閑連接，直接取最后一個元素就好，這個元素一般是最近放進去的連接。

releaseConn

func (c *HostClient) releaseConn(cc *clientConn) {
 cc.lastUseTime = time.Now()
 if c.MaxConnWaitTimeout <= 0 {
  c.connsLock.Lock()
  c.conns = append(c.conns, cc)
  c.connsLock.Unlock()
  return
 }

 // try to deliver an idle connection to a *wantConn
 c.connsLock.Lock()
 defer c.connsLock.Unlock()
 delivered := false
 if q := c.connsWait; q != nil && q.len() > 0 {
  for q.len() > 0 {
   w := q.popFront()
   if w.waiting() {
    delivered = w.tryDeliver(cc, nil)
    break
   }
  }
 }
 if !delivered {
  c.conns = append(c.conns, cc)
 }

releaseConn 會先盡量嘗試把當前的連接給正在等待連接的請求(wantConn)，彈出等待隊列(connsWait)的第一個元素。并把連接轉交給該請求。如果該請求的狀態已經不是 waiting 了，則繼續彈出，直到找到了合適的來接盤，或者等待隊列彈空。

如果沒有順利地把連接交出去，把當前連接入空閑連接數組(c.conns)。

需要注意 fasthttp 里的 conns 是連接池，clientConnPool 是 clientConn 對象的對象池。

與標準庫中的 client 不同的是，fasthttp 中沒有 read write loop，所以每個請求是在當前協程中完成的：

1. 把 request 的 header 和 body 寫入到 conn
2. 從 conn 中讀取 response
3. 釋放連接、緩存各種過程中生成的 struct 對象

gRPC

服務端

gRPC 底層基于 http2，所以交互基于 http2 stream，服務端整體流程與 http2 沒什么區別。

客戶端

在 gRPC 中，客戶端沒有使用連接池，直接使用了 http2 連接：

Invoke
-> invoke 
-> newClientStream 
-> newAttemptLocked 
-> getTransport 
-> blockingpiker.pick 
->  getReadyTransport 
->  addrConn.connect 
-> go ac.resetTransport()

然后一路走到創建 http2Client。

(dlv) bt
0  0x00000000013e2539 in google.golang.org/grpc/internal/transport.newHTTP2Client
   at /Users/xargin/go/src/google.golang.org/grpc/internal/transport/http2_client.go:167
1  0x000000000145a5ca in google.golang.org/grpc/internal/transport.NewClientTransport
   at /Users/xargin/go/src/google.golang.org/grpc/internal/transport/transport.go:575
2  0x000000000145a5ca in google.golang.org/grpc.(*addrConn).createTransport
   at /Users/xargin/go/src/google.golang.org/grpc/clientconn.go:1275
3  0x0000000001459e25 in google.golang.org/grpc.(*addrConn).tryAllAddrs
   at /Users/xargin/go/src/google.golang.org/grpc/clientconn.go:1205
4  0x00000000014593b7 in google.golang.org/grpc.(*addrConn).resetTransport
   at /Users/xargin/go/src/google.golang.org/grpc/clientconn.go:1120
5  0x000000000105b811 in runtime.goexit
   at /usr/local/go/src/runtime/asm_amd64.s:1357

thrift

thrift 官方沒有連接池，client 中生成的 seqid 只是用來和服務端返回的 rseqid 進行匹配。

func (p *TStandardClient) Recv(iprot TProtocol, seqId int32, method string, result TStruct) error {
 rMethod, rTypeId, rSeqId, err := iprot.ReadMessageBegin()
 if err != nil {
  return err
 }

 if method != rMethod {
  return NewTApplicationException(WRONG_METHOD_NAME, fmt.Sprintf("%s: wrong method name", method))
 } else if seqId != rSeqId {
  return NewTApplicationException(BAD_SEQUENCE_ID, fmt.Sprintf("%s: out of order sequence response", method))
 } else if rTypeId == EXCEPTION {
  var exception tApplicationException
  if err := exception.Read(iprot); err != nil {
   return err
  }

  if err := iprot.ReadMessageEnd(); err != nil {
   return err
  }

  return &exception
 } else if rTypeId != REPLY {
  return NewTApplicationException(INVALID_MESSAGE_TYPE_EXCEPTION, fmt.Sprintf("%s: invalid message type", method))
 }

 if err := result.Read(iprot); err != nil {
  return err
 }

 return iprot.ReadMessageEnd()
}

thrift 的每個 client 對象中包裹了一個 transport：

 ...
 useTransport, err := transportFactory.GetTransport(transport)
 client := NewEchoClientFactory(useTransport, protocolFactory)
 if err := transport.Open(); err != nil {
  fmt.Fprintln(os.Stderr, "Error opening socket to 127.0.0.1:9898", " ", err)
  os.Exit(1)
 }
 defer transport.Close()

 req := &EchoReq{Msg: "You are welcome."}
 res, err := client.Echo(context.TODO(), req)
 ...


type EchoClient struct {
 c thrift.TClient
}

func NewEchoClientFactory(t thrift.TTransport, f thrift.TProtocolFactory) *EchoClient {
 return &EchoClient{
  c: thrift.NewTStandardClient(f.GetProtocol(t), f.GetProtocol(t)),
 }
}

這個包裹的 transport 就是一條單獨的 tcp 連接，沒有連接池。

redigo

redigo 是個 client 庫，沒有服務端：

type Pool struct {
 // Dial is an application supplied function for creating and configuring a
 // connection.
 //
 // The connection returned from Dial must not be in a special state
 // (subscribed to pubsub channel, transaction started, ...).
 Dial func() (Conn, error)

 // DialContext is an application supplied function for creating and configuring a
 // connection with the given context.
 //
 // The connection returned from Dial must not be in a special state
 // (subscribed to pubsub channel, transaction started, ...).
 DialContext func(ctx context.Context) (Conn, error)

 // TestOnBorrow is an optional application supplied function for checking
 // the health of an idle connection before the connection is used again by
 // the application. Argument t is the time that the connection was returned
 // to the pool. If the function returns an error, then the connection is
 // closed.
 TestOnBorrow func(c Conn, t time.Time) error

 // Maximum number of idle connections in the pool.
 MaxIdle int

 // Maximum number of connections allocated by the pool at a given time.
 // When zero, there is no limit on the number of connections in the pool.
 MaxActive int

 // Close connections after remaining idle for this duration. If the value
 // is zero, then idle connections are not closed. Applications should set
 // the timeout to a value less than the server's timeout.
 IdleTimeout time.Duration

 // If Wait is true and the pool is at the MaxActive limit, then Get() waits
 // for a connection to be returned to the pool before returning.
 Wait bool

 // Close connections older than this duration. If the value is zero, then
 // the pool does not close connections based on age.
 MaxConnLifetime time.Duration

 chInitialized uint32 // set to 1 when field ch is initialized

 mu           sync.Mutex    // mu protects the following fields
 closed       bool          // set to true when the pool is closed.
 active       int           // the number of open connections in the pool
 ch           chan struct{} // limits open connections when p.Wait is true
 idle         idleList      // idle connections
 waitCount    int64         // total number of connections waited for.
 waitDuration time.Duration // total time waited for new connections.
}

客戶端：

redigo 的客戶端需要顯式聲明并初始化內部的 pool：

func newPool(addr string) *redis.Pool {
    return &redis.Pool{
        MaxIdle: 3,
        IdleTimeout: 240 * time.Second,
        // Dial or DialContext must be set. When both are set, DialContext takes precedence over Dial.
        Dial: func () (redis.Conn, error) { return redis.Dial("tcp", addr) },
    }
}

初始化時可以提供 TestOnBorrow 的行為：

pool := &redis.Pool{
  // Other pool configuration not shown in this example.
  TestOnBorrow: func(c redis.Conn, t time.Time) error {
    if time.Since(t) < time.Minute {
      return nil
    }
    _, err := c.Do("PING")
    return err
  },
}

使用時也需要用戶顯式地 defer Close：

func serveHome(w http.ResponseWriter, r *http.Request) {
    conn := pool.Get()
    defer conn.Close()
    ...
}

pool.Get

7

用戶需要設置 pool.Wait 是否等待，如果 Waittrue，則在沒有連接可用時，會阻塞等待。如果 Waitfalse，且連接已到達閾值 pool.MaxActive，則直接返回錯誤 ErrPoolExhausted。

activeConn.Close

func (ac *activeConn) Close() error {
 pc := ac.pc
 if pc == nil {
  return nil
 }
 ac.pc = nil

 if ac.state&connectionMultiState != 0 {
  pc.c.Send("DISCARD")
  ac.state &^= (connectionMultiState | connectionWatchState)
 } else if ac.state&connectionWatchState != 0 {
  pc.c.Send("UNWATCH")
  ac.state &^= connectionWatchState
 }
 if ac.state&connectionSubscribeState != 0 {
  pc.c.Send("UNSUBSCRIBE")
  pc.c.Send("PUNSUBSCRIBE")
  // To detect the end of the message stream, ask the server to echo
  // a sentinel value and read until we see that value.
  sentinelOnce.Do(initSentinel)
  pc.c.Send("ECHO", sentinel)
  pc.c.Flush()
  for {
   p, err := pc.c.Receive()
   if err != nil {
    break
   }
   if p, ok := p.([]byte); ok && bytes.Equal(p, sentinel) {
    ac.state &^= connectionSubscribeState
    break
   }
  }
 }
 pc.c.Do("")
 ac.p.put(pc, ac.state != 0 || pc.c.Err() != nil)
 return nil
}

close 時會把這個 activeConn 放回連接池。

go-redis/redis

https://github.com/go-redis/redis

這個 redis 庫屏蔽了連接池邏輯，用戶側基本不用關心連接，初始化時，傳入連接池相關配置：

 rdb := redis.NewClient(&redis.Options{
  Addr:     "localhost:6379", // use default Addr
  Password: "",               // no password set
  DB:       0,                // use default DB
 })

func NewClient(opt *Options) *Client {
 opt.init()

 c := Client{
  baseClient: newBaseClient(opt, newConnPool(opt)),
  ctx:        context.Background(),
 }
 c.cmdable = c.Process

 return &c
}

func newConnPool(opt *Options) *pool.ConnPool {
 return pool.NewConnPool(&pool.Options{
  Dialer: func(ctx context.Context) (net.Conn, error) {
   return opt.Dialer(ctx, opt.Network, opt.Addr)
  },
  PoolSize:           opt.PoolSize,
  MinIdleConns:       opt.MinIdleConns,
  MaxConnAge:         opt.MaxConnAge,
  PoolTimeout:        opt.PoolTimeout,
  IdleTimeout:        opt.IdleTimeout,
  IdleCheckFrequency: opt.IdleCheckFrequency,
 })
}
func (c *baseClient) _process(ctx context.Context, cmd Cmder) error {
 var lastErr error
 for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
  if attempt > 0 {
   if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
    return err
   }
  }

  retryTimeout := true
  lastErr = c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
   err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
    return writeCmd(wr, cmd)
   })
   if err != nil {
    return err
   }

   err = cn.WithReader(ctx, c.cmdTimeout(cmd), cmd.readReply)
   if err != nil {
    retryTimeout = cmd.readTimeout() == nil
    return err
   }

   return nil
  })
  if lastErr == nil || !isRetryableError(lastErr, retryTimeout) {
   return lastErr
  }
 }
 return lastErr
}

func (c *baseClient) withConn(
 ctx context.Context, fn func(context.Context, *pool.Conn) error,
) error {
 cn, err := c.getConn(ctx)
 if err != nil {
  return err
 }
 defer func() {
  c.releaseConn(cn, err)
 }()

 err = fn(ctx, cn)
 return err

連接池維護的邏輯和其它庫差不多。與其它庫不同的是，該庫會保證 idle 的 conns 維持在 MinIdleConn 配置數量之上，不足的話，會在后臺補充：

func (p *ConnPool) checkMinIdleConns() {
 if p.opt.MinIdleConns == 0 {
  return
 }
 for p.poolSize < p.opt.PoolSize && p.idleConnsLen < p.opt.MinIdleConns {
  p.poolSize++
  p.idleConnsLen++
  go func() {
   err := p.addIdleConn()
   if err != nil {
    p.connsMu.Lock()
    p.poolSize--
    p.idleConnsLen--
    p.connsMu.Unlock()
   }
  }()
 }
}

database/sql

這里的連接池與 RPC 系列的稍有區別，取的是 freeConns 的第一個，并且有一個可能效率比較低的 copy 過程：

 // Prefer a free connection, if possible.
 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
  }
  // Lock around reading lastErr to ensure the session resetter finished.
  conn.Lock()
  err := conn.lastErr
  conn.Unlock()
  if err == driver.ErrBadConn {
   conn.Close()
   return nil, driver.ErrBadConn
  }
  return conn, nil
 

其它的沒啥特殊的。