Go創建Grpc鏈接池
轉載請注明來源：https://janrs.com/2023/03/創建grpc鏈接池/
常規用法
gRPC 四種基本使用
- 請求回應模式
- 客戶端資料流模式
- 服務端資料流模式
- 雙向流模式
常見的gRPC呼叫寫法
func main(){
//... some code
// 鏈接grpc服務
conn , err := grpc.Dial(":8000",grpc.WithInsecure)
if err != nil {
//...log
}
defer conn.Close()
//...some code
存在的問題：面臨高并發的情況，性能問題很容易就會出現，例如我們在做性能測驗的時候，就會發現，打一會性能測驗，客戶端請求服務端的時候就會報錯：
rpc error: code = Unavailable desc = all SubConns are in TransientFailure, latest connection error: connection error: desc = "transport: Error while dialing dial tcp xxx:xxx: connect: connection refused
實際去查看問題的時候，很明顯，這是 gRPC 的連接數被打滿了，很多連接都還未完全釋放，[#本文來源：janrs.com#]
gRPC 的通信本質上也是 TCP 的連接，那么一次連接就需要三次握手，和四次揮手，每一次建立連接和釋放連接的時候，都需要走這么一個程序，如果我們頻繁的建立和釋放連接，這對于資源和性能其實都是一個大大的浪費，
在服務端，gRPC 服務端的鏈接管理不用我們操心，但是 gRPC 客戶端的鏈接管理非常有必要關心，要實作復用客戶端的連接，
創建鏈接池
創建鏈接池需要考慮的問題：
- 連接池是否支持擴縮容
- 空閑的連接是否支持超時自行關閉，是否支持保活
- 池子滿的時候，處理的策略是什么樣的
創建鏈接池介面
type Pool interface {
// 獲取一個新的連接 ， 當關閉連接的時候，會將該連接放入到池子中
Get() (Conn, error)
// 關閉連接池，自然連接池子中的連接也不再可用
Close() error
//[#本文來源：janrs.com#]
Status() string
}
實作鏈接池介面
創建鏈接池代碼
func New(address string, option Options) (Pool, error) {
if address == "" {
return nil, errors.New("invalid address settings")
}
if option.Dial == nil {
return nil, errors.New("invalid dial settings")
}
if option.MaxIdle <= 0 || option.MaxActive <= 0 || option.MaxIdle > option.MaxActive {
return nil, errors.New("invalid maximum settings")
}
if option.MaxConcurrentStreams <= 0 {
return nil, errors.New("invalid maximun settings")
}
p := &pool{
index: 0,
current: int32(option.MaxIdle),
ref: 0,
opt: option,
conns: make([]*conn, option.MaxActive),
address: address,
closed: 0,
}
for i := 0; i < p.opt.MaxIdle; i++ {
c, err := p.opt.Dial(address)
if err != nil {
p.Close()
return nil, fmt.Errorf("dial is not able to fill the pool: %s", err)
}
p.conns[i] = p.wrapConn(c, false)
}
log.Printf("new pool success: %v\n", p.Status())
return p, nil
}
關于以上的代碼，需要特別注意每一個連接的建立也是在 New 里面完成的，[#本文來源：janrs.com#]只要有 1 個連接未建立成功，那么咱們的連接池就算是建立失敗，咱們會呼叫 p.Close() 將之前建立好的連接全部釋放掉，
關閉鏈接池代碼
// 關閉連接池
func (p *pool) Close() error {
atomic.StoreInt32(&p.closed, 1)
atomic.StoreUint32(&p.index, 0)
atomic.StoreInt32(&p.current, 0)
atomic.StoreInt32(&p.ref, 0)
p.deleteFrom(0)
log.Printf("[janrs.com]close pool success: %v\n", p.Status())
return nil
}
從具體位置洗掉鏈接池代碼
// 清除從 指定位置開始到 MaxActive 之間的連接
func (p *pool) deleteFrom(begin int) {
for i := begin; i < p.opt.MaxActive; i++ {
p.reset(i)
}
}
銷毀具體的鏈接代碼
// 清除具體的連接
func (p *pool) reset(index int) {
conn := p.conns[index]
if conn == nil {
return
}
conn.reset()
p.conns[index] = nil
}
關閉鏈接
代碼
func (c *conn) reset() error {
cc := c.cc
c.cc = nil
c.once = false
// 本文博客來源：janrs.com
if cc != nil {
return cc.Close()
}
return nil
}
func (c *conn) Close() error {
c.pool.decrRef()
if c.once {
return c.reset()
}
return nil
}
在使用連接池通過 pool.Get() 拿到具體的連接句柄 conn 之后，會使用 conn.Close()關閉連接，實際上也是會走到上述的 Close() 實作的位置，但是并未指定當然也沒有權限顯示的指定將 once 置位為 false ，也就是對于呼叫者來說，是關閉了連接，對于連接池來說，實際上是將連接歸還到連接池中，
擴縮容
關鍵代碼
func (p *pool) Get() (Conn, error) {
// the first selected from the created connections
nextRef := p.incrRef()
p.RLock()
current := atomic.LoadInt32(&p.current)
p.RUnlock()
if current == 0 {
return nil, ErrClosed
}
if nextRef <= current*int32(p.opt.MaxConcurrentStreams) {
next := atomic.AddUint32(&p.index, 1) % uint32(current)
return p.conns[next], nil
}
// 本文博客來源：janrs.com
// the number connection of pool is reach to max active
if current == int32(p.opt.MaxActive) {
// the second if reuse is true, select from pool's connections
if p.opt.Reuse {
next := atomic.AddUint32(&p.index, 1) % uint32(current)
return p.conns[next], nil
}
// the third create one-time connection
c, err := p.opt.Dial(p.address)
return p.wrapConn(c, true), err
}
// the fourth create new connections given back to pool
p.Lock()
current = atomic.LoadInt32(&p.current)
if current < int32(p.opt.MaxActive) && nextRef > current*int32(p.opt.MaxConcurrentStreams) {
// 2 times the incremental or the remain incremental ##janrs.com
increment := current
if current+increment > int32(p.opt.MaxActive) {
increment = int32(p.opt.MaxActive) - current
}
var i int32
var err error
for i = 0; i < increment; i++ {
c, er := p.opt.Dial(p.address)
if er != nil {
err = er
break
}
p.reset(int(current + i))
p.conns[current+i] = p.wrapConn(c, false)
}
// 本文博客來源：janrs.com
current += i
log.Printf("#janrs.com#grow pool: %d ---> %d, increment: %d, maxActive: %d\n",
p.current, current, increment, p.opt.MaxActive)
atomic.StoreInt32(&p.current, current)
if err != nil {
p.Unlock()
return nil, err
}
}
p.Unlock()
next := atomic.AddUint32(&p.index, 1) % uint32(current)
return p.conns[next], nil
}
Get 代碼邏輯
- 先增加連接的參考計數，如果在設定
current*int32(p.opt.MaxConcurrentStreams)范圍內，那么直接取連接進行使用即可， - 若當前的連接數達到了最大活躍的連接數，那么就看我們新建池子的時候傳遞的
option中的reuse引數是否是true，若是復用，則隨機取出連接池中的任意連接提供使用，如果不復用，則新建一個連接， - 其余的情況，就需要我們進行
2倍或者1倍的數量對連接池進行擴容了，
也可以在 Get 的實作上進行縮容，具體的縮容策略可以根據實際情況來定，例如當參考計數 nextRef 只有當前活躍連接數的 10% 的時候（這只是一個例子），就可以考慮縮容了，
性能測驗
有關鏈接池的創建以及性能測驗
https://mycodesmells.com/post/pooling-grpc-connections
轉載請注明來源：https://janrs.com/2023/03/創建grpc鏈接池/
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/545548.html
標籤：其他