go的pool资源池:
1.当有多个并发请求的时候,比如需要查询数据库2.先创建一个2个容量的数据库连接资源池3.当一个请求过来的时候,去资源池里请求连接资源,肯定是空的就创建一个连接,执行查询,结束后放入了资源池里4.当第二个请求过来的时候,也是去资源池请求连接资源,就直接在池中拿过来一个连接进行查询5.当并发大的时候,资源池里面没有足够连接资源,就会不停创建新资源,放入池里面的时候,也会放不进去,就主动关闭掉这个资源6.这里的资源池实质上是一个缓冲通道,里面放着连接资源
package mainimport ( "errors" "io" "log" "math/rand" "sync" "sync/atomic" "time")//定义一个结构体,这个实体类型可以作为整体单元被复制,可以作为参数或返回值,或被存储到数组type Pool struct { //定义成员,互斥锁类型 m sync.Mutex //定义成员,通道类型,通道传递的是io.Closer类型 resources chan io.Closer //定义工厂成员,类型是func()(io.Closer,error) //error是预定义类型,实际上是个interface接口类型 factory func() (io.Closer, error) closed bool}//定义变量,函数返回的是error类型var ErrPoolClosed = errors.New("池已经关闭了")//定义New方法,创建一个池,返回的是Pool类型的指针//传入的参数是个函数类型func(io.Closer,error)和池的大小func New(fn func() (io.Closer, error), size uint) (*Pool, error) { //使用结构体字面值给结构体成员赋值 myPool := Pool{ factory: fn, resources: make(chan io.Closer, size), } //返回两个返回值 return &myPool, nil}//从池中请求获取一个资源,给Pool类型定义的方法//返回的值是io.Closer类型func (p *Pool) Acquire() (io.Closer, error) { //基于select的多路复用 //select会等待case中有能够执行的,才会去执行,等待其中一个能执行就执行 //default分支会在所有case没法执行时,默认执行,也叫轮询channel select { case r, _ := <-p.resources: log.Printf("请求资源:来自通道 %d", r.(*dbConn).ID) return r, nil //如果缓冲通道中没有了,就会执行这里 default: log.Printf("请求资源:创建新资源") return p.factory() }}//将一个使用后的资源放回池//传入的参数是io.Closer类型func (p *Pool) Release(r io.Closer) { //使用mutex互斥锁 p.m.Lock() //解锁 defer p.m.Unlock() //如果池都关闭了 if p.closed { //关掉资源 r.Close() return } //select多路选择 //如果放回通道的时候满了,就关闭这个资源 select { case p.resources <- r: log.Printf("释放资源:放入通道 %d", r.(*dbConn).ID) default: log.Printf("释放资源:关闭资源%d", r.(*dbConn).ID) r.Close() }}//关闭资源池,关闭通道,将通道中的资源关掉func (p *Pool) Close() { p.m.Lock() defer p.m.Unlock() p.closed = true //先关闭通道再清空资源 close(p.resources) //清空并关闭资源 for r := range p.resources { r.Close() }}//定义全局常量const ( maxGoroutines = 20 //使用25个goroutine模拟同时的连接请求 poolSize = 2 //资源池中的大小)//定义结构体,模拟要共享的资源type dbConn struct { //定义成员 ID int32}//dbConn实现io.Closer接口func (db *dbConn) Close() error { return nil}var idCounter int32 //定义一个全局的共享的变量,更新时用原子函数锁住//定义方法,创建dbConn实例//返回的是io.Closer类型和error类型func createConn() (io.Closer, error) { //原子函数锁住,更新加1 id := atomic.AddInt32(&idCounter, 1) log.Printf("创建新资源: %d", id) return &dbConn{id}, nil}func main() { //计数信号量 var wg sync.WaitGroup //同时并发的数量 wg.Add(maxGoroutines) myPool, _ := New(createConn, poolSize) //开25个goroutine同时查询 for i := 0; i < maxGoroutines; i++ { //模拟请求 time.Sleep(time.Duration(rand.Intn(2)) * time.Second) go func(gid int) { execQuery(gid, myPool) wg.Done() }(i) } //等待上面开的goroutine结束 wg.Wait() myPool.Close()}//定义一个查询方法,参数是当前gorotineId和资源池func execQuery(goroutineId int, pool *Pool) { //从池里请求资源,第一次肯定是没有的,就会创建一个dbConn实例 conn, _ := pool.Acquire() //将创建的dbConn实例放入了资源池的缓冲通道里 defer pool.Release(conn) //睡眠一下,模拟查询过程 time.Sleep(time.Duration(rand.Intn(10)) * time.Second) log.Printf("执行查询...协程ID [%d] 资源ID [%d]", goroutineId, conn.(*dbConn).ID)} 