golang实现基于channel的通用连接池详解

时间:2022-09-22 13:23:21

前言

golang的channel除了goroutine通信之外还有很多其他的功能,本文将实现一种基于channel的通用连接池。下面话不多说了,来一起看看详细的介绍吧。

功能

* 连接池中连接类型为interface{},使得更加通用

* 链接的最大空闲时间,超时的链接将关闭丢弃,可避免空闲时链接自动失效问题

* 使用channel处理池中的链接,高效

何为通用?

连接池的实现不依赖具体的实例,而依赖某个接口,本文的连接池选用的是io.Closer接口,只要是实现了该接口的对象都可以被池管理。

当然,你可以实现基于interface{}的连接池,这样任何对象都可以被管理。

实现原理

将连接句柄存入channel中,由于缓存channel的特性,获取连接时如果池中有连接,将直接返回,如果池中没有连接,将阻塞或者新建连接(没超过最大限制的情况下)。

由于面向接口编程,所有创建连接的逻辑是不清楚的,这里需要传入一个函数,该函数返回一个io.Closer对象。

实现

由于并发问题,在需要操作池中互斥数据的时候需要加锁。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
package pool
import (
  "errors"
  "io"
  "sync"
  "time"
)
 
var (
  ErrInvalidConfig = errors.New("invalid pool config")
  ErrPoolClosed  = errors.New("pool closed")
)
 
type factory func() (io.Closer, error)
 
type Pool interface {
  Acquire() (io.Closer, error) // 获取资源
  Release(io.Closer) error   // 释放资源
  Close(io.Closer) error    // 关闭资源
  Shutdown() error       // 关闭池
}
 
type GenericPool struct {
  sync.Mutex
  pool    chan io.Closer
  maxOpen   int // 池中最大资源数
  numOpen   int // 当前池中资源数
  minOpen   int // 池中最少资源数
  closed   bool // 池是否已关闭
  maxLifetime time.Duration
  factory   factory // 创建连接的方法
}
 
func NewGenericPool(minOpen, maxOpen int, maxLifetime time.Duration, factory factory) (*GenericPool, error) {
  if maxOpen <= 0 || minOpen > maxOpen {
    return nil, ErrInvalidConfig
  }
  p := &GenericPool{
    maxOpen:   maxOpen,
    minOpen:   minOpen,
    maxLifetime: maxLifetime,
    factory:   factory,
    pool:    make(chan io.Closer, maxOpen),
  }
 
  for i := 0; i < minOpen; i++ {
    closer, err := factory()
    if err != nil {
      continue
    }
    p.numOpen++
    p.pool <- closer
  }
  return p, nil
}
 
func (p *GenericPool) Acquire() (io.Closer, error) {
  if p.closed {
    return nil, ErrPoolClosed
  }
  for {
    closer, err := p.getOrCreate()
    if err != nil {
      return nil, err
    }
    // todo maxLifttime处理
    return closer, nil
  }
}
 
func (p *GenericPool) getOrCreate() (io.Closer, error) {
  select {
  case closer := <-p.pool:
    return closer, nil
  default:
  }
  p.Lock()
  if p.numOpen >= p.maxOpen {
    closer := <-p.pool
    p.Unlock()
    return closer, nil
  }
  // 新建连接
  closer, err := p.factory()
  if err != nil {
    p.Unlock()
    return nil, err
  }
  p.numOpen++
  p.Unlock()
  return closer, nil
}
 
// 释放单个资源到连接池
func (p *GenericPool) Release(closer io.Closer) error {
  if p.closed {
    return ErrPoolClosed
  }
  p.Lock()
  p.pool <- closer
  p.Unlock()
  return nil
}
 
// 关闭单个资源
func (p *GenericPool) Close(closer io.Closer) error {
  p.Lock()
  closer.Close()
  p.numOpen--
  p.Unlock()
  return nil
}
 
// 关闭连接池,释放所有资源
func (p *GenericPool) Shutdown() error {
  if p.closed {
    return ErrPoolClosed
  }
  p.Lock()
  close(p.pool)
  for closer := range p.pool {
    closer.Close()
    p.numOpen--
  }
  p.closed = true
  p.Unlock()
  return nil
}

结论

基于该连接池,可以管理所有io.Closer对象。比如memcached,redis等等,非常方便!

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对服务器之家的支持。

原文链接:https://segmentfault.com/a/1190000013089363