go内存池Pool

1 go提供的sync.Pool是为了对象的复用，如果某些对象的创建比较频繁，就把他们放入Pool中缓存起来以便使用，这样重复利用内存，减少GC的压力,

package main
import (
    "errors"
    "io"
    "log"
    "math/rand"
    "sync"
    "sync/atomic"
    "time"
    //"flysnow.org/hello/common"
)
//一个安全的资源池，被管理的资源必须都实现io.Close接口
type Pool struct {
    m       sync.Mutex
    res     chan io.Closer
    factory func() (io.Closer, error)
    closed  bool
}
var ErrPoolClosed = errors.New("资源池已经被关闭。")
//创建一个资源池
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
    if size <= 0 {
        return nil, errors.New("size的值太小了。")
    }
    return &Pool{
        factory: fn,
        res:     make(chan io.Closer, size),
    }, nil
}
//从资源池里获取一个资源
func (p *Pool) Acquire() (io.Closer, error) {
    select {
    case r, ok := <-p.res:
        log.Println("Acquire:共享资源")
        if !ok {
            return nil, ErrPoolClosed
        }
        return r, nil
    default:
        log.Println("Acquire:新生成资源")
        return p.factory()
    }
}
//关闭资源池，释放资源
func (p *Pool) Close() {
    p.m.Lock()
    defer p.m.Unlock()
    if p.closed {
        return
    }
    p.closed = true
    //关闭通道，不让写入了
    close(p.res)
    //关闭通道里的资源
    for r := range p.res {
        r.Close()
    }
}
func (p *Pool) Release(r io.Closer) {
    //保证该操作和Close方法的操作是安全的
    p.m.Lock()
    defer p.m.Unlock()
    //资源池都关闭了，就省这一个没有释放的资源了，释放即可
    if p.closed {
        r.Close()
        return
    }
    select {
    case p.res <- r:
        log.Println("资源释放到池子里了")
    default:
        log.Println("资源池满了，释放这个资源吧")
        r.Close()
    }
}
const (
    //模拟的最大goroutine
    maxGoroutine = 5
    //资源池的大小
    poolRes      = 2
)
func main() {
    //等待任务完成
    var wg sync.WaitGroup
    wg.Add(maxGoroutine)
    p, err := New(createConnection, poolRes)
    if err != nil {
        log.Println(err)
        return
    }
    // 模拟好几个goroutine同时使用资源池查询数据
    for query := 0; query < maxGoroutine; query++ {
        go func(q int) {
            dbQuery(q, p)
            wg.Done()
        }(query)
    }
    wg.Wait()
    log.Println("开始关闭资源池")
    p.Close()
}
//模拟数据库查询
func dbQuery(query int, pool *Pool) {
    conn, err := pool.Acquire()
    if err != nil {
        log.Println(err)
        return
    }
    // 防止忘记释放资源，
    defer pool.Release(conn)
    //模拟查询
    time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
    log.Printf("第%d个查询，使用的是ID为%d的数据库连接", query, conn.(*dbConnection).ID)
}
//数据库连接
type dbConnection struct {
    ID int32//连接的标志
}
// 实现io.Closer接口
func (db *dbConnection) Close() error {
    log.Println("关闭连接", db.ID)
    return nil
}
var idCounter int32
// 生成数据库连接的方法，以供资源池使用，这个函数符合Pool中的factory的类型，
func createConnection() (io.Closer, error) {
    //并发安全，给数据库连接生成唯一标志
    id := atomic.AddInt32(&idCounter, 1)
    return &dbConnection{id}, nil
}

View Code

https://www.flysnow.org/2017/05/01/go-in-action-go-pool.html

Pool创建的时候是不能指定大小的，所有sync.Pool的缓存对象数量是没有限制的（只受限于内存），当执行一个pool的get或者put操作的时候都会先把当前的goroutine固定到某个P的子池上面，然后再对该子池进行操作。每个子池里面有一个私有对象和共享列表对象，私有对象是只有对应的P能够访问，因为一个P同一时间只能执行一个goroutine，因此对私有对象存取操作是不需要加锁的。共享列表是和其他P分享的，因此操作共享列表是需要加锁的。

Get的过程是先找当前P的子池里的私有对象，如果有返回，如果为空去当前的子池的共享列表里去找（要加锁），如果还没有，去其它P的共享列表里去找（要加锁），如果还没有，去New一个，一次get操作最少0次加锁，最大N（N等于MAXPROCS）次加锁，

Put的过程是先固定到某个P，如果私有对象为空则放到私有对象；否则加入到该P子池的共享列表中（要加锁），一次put操作最少0次加锁，最多1次加锁。sync.Pool的定位不是做类似连接池的东西，它的用途仅仅是增加对象重用的几率，减少gc的负担，而开销方面也不是很便宜的

https://blog.csdn.net/yongjian_lian/article/details/42058893

https://studygolang.com/articles/21023

1 写 go 并发程序的时候如果程序会启动大量的 goroutine ，势必会消耗大量的系统资源（内存，CPU），实例化一个协程池，可以复用 goroutine ，节省资源，

用go实现一个协程池，为了解耦合，所以用了内外两个channel，这样协程池内部的方法不会暴露到外面，

package main
import (
    "fmt"
    "time"
)
type Task struct {
    f func() error
}
func NewTask(arg_f func() error) *Task {
    t := Task{
        f: arg_f,
    }
    return &t
}
func (t *Task) Execute() {
    t.f()
}
type Pool struct {
    EntryChannel chan *Task
    JobsChannel  chan *Task
    worker_num   int
}
func NewPool(cap int) *Pool {
    p := Pool{
        EntryChannel: make(chan *Task),
        JobsChannel:  make(chan *Task),
        worker_num:   cap,
    }
    return &p
}
func (p *Pool) worker(worker_ID int) {
    // 一旦工作channel中有task了，就取出来执行，
    for task := range p.JobsChannel {
        task.Execute()
        fmt.Println("worker ID", worker_ID, " 执行完了一个任务")
    }
}
func (p *Pool) run() {
    // 启动了4个goroutine来执行Pool中的任务，
    for i := 0; i < p.worker_num; i++ {
        go p.worker(i)
    }
    // 从进入channel中取出来，放入工作channel，
    for task := range p.EntryChannel {
        p.JobsChannel <- task
    }
}
func main() {
    t := NewTask(func() error {
        fmt.Println(time.Now())
        fmt.Print("这是处理Task的函数，")
        return nil
    })
    p := NewPool(4)
    task_num := 0
    go func() {
        for {
            p.EntryChannel <- t
            task_num += 1
            fmt.Print("当前一共执行了", task_num, " 个任务")
        }
    }()
    p.run()
}

View Code

https://www.bilibili.com/video/BV1GV411S7Ra/?spm_id_from=333.788.recommend_more_video.1