如何制作产品网站模板下载,python和c++学哪个好,邢台市天气预报15天,wordpress的polylanggo Pool
Pool用于展示如何使用有缓冲的通道实现资源池#xff0c;来管理可以在任意数量的goroutine之间共享及独立使用的资源。这种模式在需要共享一组静态资源的情况#xff08;如共享数据库连接或者内存缓冲区#xff09;下非 常有用。如果goroutine需要从池里得到这些资…go Pool
Pool用于展示如何使用有缓冲的通道实现资源池来管理可以在任意数量的goroutine之间共享及独立使用的资源。这种模式在需要共享一组静态资源的情况如共享数据库连接或者内存缓冲区下非 常有用。如果goroutine需要从池里得到这些资源中的一个它可以从池里申请使用完后归还到资源池里。
按照上面的要求整个资源池的要求和框架图梳理如下
资源池能够根据工厂函数创建多个也就是资源池要是个通用的不能只针对一种资源的资源池资源池要有上限资源池需要保证多线程安全资源池在回收资源时如果资源已经超过资源池上限不能放到资源池里面需要直接释放Close掉
提到池子channel天然就是一个池子而且channel是多线程安全的所以可以直接使用管道来安全的共享资源。文章按照如下四个方面对Pool资源池的设计过程以及使用进行剖析。
数据结构
既然确定了使用管道进行资源的共享那么Pool里面就不能少了管道。
resources chan io.Closer这里我们定义一个类型为io.Closer的chan因为管道中的资源都需要再不使用时清理资源因此这里所有传入到管道的资源都需要实现io.Closer接口。
向管道中放入资源和从管道中取出资源是多线程安全的但是打开管道和关闭管道本身又是线程不安全的因此这里需要对管道的创建和关闭进行保护 这里采用sync.Mutex互斥锁进行保护。
mutex sync.Mutex除了管道Pool还需要一个资源创建函数用于在资源池里创建资源。
factory func() (io.Closer, error)factory用于创建资源返回一个实现了io.Closer的对象表示创建的资源。
最后还需要一个原子变量用来表示Pool的状态。
closed boolatomic包提供了一些原子操作包括对bool变量的设置。
错误处理
当向一个已经关闭的Pool请求资源时需要返回一个错误提醒用户该Pool已经关闭。
var ErrPoolClosed errors.New(Pool has been closed.)方法实现
创建Pool
调用New方法创建Pool需要提供一个资源创建函数以及资源池中初始化的资源数量。
// New 创建一个新的池子这个池子会调用factory创建资源并可设置资源上限
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {if size 0 {return nil, errors.New(size value too small)}return Pool{factory: fn,resources: make(chan io.Closer, size),}, nil
}获取资源
既然是资源池那么就要提供获取资源的方法这里采用Acquire作为获取资源的方法。资源获取时有两种情况
如果有空闲资源就直接返回该资源。如果没有空闲资源就调用factory创建一个新资源。 // Acquire 从池子中获取一个资源
func (p *Pool) Acquire() (io.Closer, error) {select {// 检查是否有空闲资源case r, ok : -p.resources:if !ok {return nil, ErrPoolClosed}return r, nil// 因为没有空闲资源所以创建一个新资源default:if p.factory nil {return nil, ErrPoolClosed}return p.factory()}
}这里用来select的一个技巧将阻塞的资源变成非阻塞的分支。当向一个空管道中获取资源时会阻塞住但是通过select可以在通道阻塞的情况下执行default分支从而将阻塞的操作转换为非阻塞。
释放资源
当释放资源时也存在两种情况
如果资源池已经关闭就直接关闭该资源。如果资源池没有满就直接将资源放回资源池。
// Release 将一个使用后的资源放回池子里面
func (p *Pool) Release(r io.Closer) {// Secure this operation with the Close operation.p.m.Lock()defer p.m.Unlock()// If the pool is closed, discard the resource.if p.closed {err : r.Close()if err ! nil {return}return}// select能将所有阻塞的资源通过默认分支变成非阻塞的分支select {// 试图将资源放回资源池case p.resources - r:log.Println(Release:, In Queue)// 如果资源池已经满了就关闭这个资源default:log.Println(Release:, Closing)err : r.Close()if err ! nil {return}}
}关闭资源池Pool
当资源池使用完毕时需要及时关闭资源池并释放掉所有的资源。
// Close 关闭资源池
func (p *Pool) Close() {// 保证多线程安全p.m.Lock()defer p.m.Unlock()// 如果资源池已经关闭就不用再关闭了if p.closed {return}// 关闭资源池并释放所有资源close(p.resources)p.closed true// 通道即使关闭了也能通过range将通道中的资源全部接收一遍for r : range p.resources {err : r.Close()if err ! nil {return}}
}按照上述步骤将资源池的视线示例汇总如下
// Package pool 管理用户资源的池子
package poolimport (errorsiologsync
)// 说明go 1.6之后自带sync.Pool用来管理资源// Pool 管理用户资源的池子可以在多个goroutine之间共享资源
type Pool struct {m sync.Mutexresources chan io.Closerfactory func() (io.Closer, error)closed bool
}// ErrPoolClosed 表示(Acquire)了一个已经关闭的池子
var ErrPoolClosed errors.New(pool has been closed)// New 创建一个新的池子这个池子会调用factory创建资源并可设置资源上限
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {if size 0 {return nil, errors.New(size value too small)}return Pool{factory: fn,resources: make(chan io.Closer, size),}, nil
}// Acquire 从池子中获取一个资源
func (p *Pool) Acquire() (io.Closer, error) {select {// 检查是否有空闲资源case r, ok : -p.resources:if !ok {return nil, ErrPoolClosed}return r, nil// 因为没有空闲资源所以创建一个新资源default:if p.factory nil {return nil, ErrPoolClosed}return p.factory()}
}// Release 将一个使用后的资源放回池子里面
func (p *Pool) Release(r io.Closer) {// Secure this operation with the Close operation.p.m.Lock()defer p.m.Unlock()// If the pool is closed, discard the resource.if p.closed {err : r.Close()if err ! nil {return}return}// select能将所有阻塞的资源通过默认分支变成非阻塞的分支select {// 试图将资源放回资源池case p.resources - r:log.Println(Release:, In Queue)// 如果资源池已经满了就关闭这个资源default:log.Println(Release:, Closing)err : r.Close()if err ! nil {return}}
}// Close 关闭资源池
func (p *Pool) Close() {// 保证多线程安全p.m.Lock()defer p.m.Unlock()// 如果资源池已经关闭就不用再关闭了if p.closed {return}// 关闭资源池并释放所有资源close(p.resources)p.closed true// 通道即使关闭了也能通过range将通道中的资源全部接收一遍for r : range p.resources {err : r.Close()if err ! nil {return}}
}调用示例
// 这个示例展示了Pool的使用其实新版本的go里面已经内置了资源池这里只是为了演示如何使用自己实现的Pool
package mainimport (iologmath/randsyncsync/atomictimepatterns/pool
)const (maxGoroutines 25 // 协程数量.pooledResources 2 // 资源池的大小
)// 数据库连接池资源信息.
type dbConnection struct {ID int32
}// Close 实现 io.Closer 接口
func (dbConn *dbConnection) Close() error {log.Println(Close: Connection, dbConn.ID)return nil
}// idCounter 为每个连接生成一个唯一的ID
var idCounter int32// createConnection 连接创建工厂函数
func createConnection() (io.Closer, error) {id : atomic.AddInt32(idCounter, 1)log.Println(Create: New Connection, id)return dbConnection{id}, nil
}func main() {// 协程同步var wg sync.WaitGroupwg.Add(maxGoroutines)// 创建资源池.p, err : pool.New(createConnection, pooledResources)if err ! nil {log.Println(err)}// 资源池性能模拟测试.for query : 0; query maxGoroutines; query {// 这里将query作为参数传入是为了避免闭包时引用导致所有协程都拿到的是同一个query的值go func(q int) {performQueries(q, p)wg.Done()}(query)}// 等待所有协程结束wg.Wait()// 关闭资源池log.Println(Shutdown Program.)p.Close()
}// performQueries 测试资源池的查询表现
func performQueries(query int, p *pool.Pool) {// 获取一个连接池连接conn, err : p.Acquire()if err ! nil {log.Println(err)return}// 将连接信息放回连接池defer p.Release(conn)// 使用延迟模拟查询事件time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)log.Printf(Query: QID[%d] CID[%d]\n, query, conn.(*dbConnection).ID)
}附录
数据来源-《go语言实战》代码仓库gitee pool 如果感觉文章对你有用欢迎点赞评论和关注谢谢
