Golang代码搜集-资源池管理-蒲公英云

Golang代码搜集-资源池管理

忘是亡心i 2021-09-14 13:32 478阅读 0赞

这段代码来自《Go语言实战》第7.2节，演示如何编写一个资源池，可用于数据库连接等。原代码资源管理的效果并不明显，笔者对测试用代码进行了修改，使效果更明显。下面直接上代码。

//资源池
package pool
import (
    "errors"
    "io"
    "log"
    "sync"
)
// Pool 管理一组可以安全地在多个goroutine间共享的资源
// 被共享的资源必须实现 io.Closer接口
type Pool struct {
    m         sync.Mutex
    resources chan io.Closer
    factory   func() (io.Closer, error)
    closed    bool
}
// 资源池已经关闭
var ErrPoolClosed = errors.New("Pool has been closed.")
// 输入资源池size太小
var ErrPoolSizeTooSmall = errors.New("Pool size too small")
// New创建一个用来管事的资源池
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
    if size <= 0 {
        return nil, ErrPoolSizeTooSmall
    }
    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
        }
        log.Println("Acquire:", " Shared Resource")
        return r, nil
    default:
        //没有空闲资源，提供新资源
        log.Println("Acquire:", " New Resource")
        return p.factory()
    }
}
func (p *Pool) Release(r io.Closer) {
    //保证本操作和close操作的安全
    p.m.Lock()
    defer p.m.Unlock()
    //如果资源池已经关闭，则关闭当前资源
    if p.closed {
        r.Close()
        return
    }
    select {
    //试图将这资源放入队列
    case p.resources <- r:
        log.Println("Release:", "In Queue")
    default:
        //如果队列已满，则关闭这个资源
        log.Println("Release:", "Closing")
        r.Close()
    }
}
//Close 资源池停止工作，并关闭所有现有的资源
func (p *Pool) Close() {
    //保证本操作和Release操作的安全
    p.m.Lock()
    defer p.m.Unlock()
    //如果已经关闭，则什么也不做
    if p.closed {
        return
    }
    //关闭资源池
    p.closed = true
    //关闭通道,必须先关闭
    close(p.resources)
    //关闭资源
    for r := range p.resources {
        r.Close()
    }
}
//使用示例
package main
import (
    "fmt"
    "io"
    "log"
    "math/rand"
    "pool"
    "sync"
    "sync/atomic"
    "time"
)
var (
    maxGoroutines        = 10
    pooledResources uint = 5
)
// dbConnection 模拟共享资源
type dbConnection struct {
    Id int32
}
// 实现io.Closer接口
func (db *dbConnection) Close() error {
    log.Println("Close:Connection", db.Id)
    return nil
}
// 用来给第个链接一个独一无二的id
var idCounter int32
func createConn() (io.Closer, error) {
    id := atomic.AddInt32(&idCounter, 1)
    log.Println("Create:New Conn", id)
    return &dbConnection{id}, nil
}
func main() {
    rand.Seed(time.Now().UnixNano())
    fmt.Println("This demo presents how to use the pool")
    //创建资源池
    p, err := pool.New(createConn, pooledResources)
    if err != nil {
        log.Fatal("Create pool failed")
    }
    //关闭资源池
    defer p.Close()
    var wg sync.WaitGroup
    wg.Add(maxGoroutines)
    for query := 0; query < maxGoroutines; query++ {
        go performQueries(query, p, &wg)
        //随机等一段时间，否则后而gorountine在获取资源时
        //资源池仍没有资源
        time.Sleep(time.Duration(rand.Intn(2)) * time.Second)
    }
    //等待 结束
    wg.Wait()
    fmt.Printf("Query num:%d, Connection num:%d\n", maxGoroutines, idCounter)
}
func performQueries(query int, p *pool.Pool, wg *sync.WaitGroup) {
    defer wg.Done()
    //从池里请求一个链接
    conn, err := p.Acquire()
    if err != nil {
        log.Fatal(err)
    }
    defer p.Release(conn)
    //模拟查询
    time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
    log.Printf("QID[%d] CID[%d]\n", query, conn.(*dbConnection).Id)
}