聊一聊Go网络编程(二)--TCP连接管理

func ForWhileSignal(sig *signal, gid string, mutx *sync.Mutex, cd *sync.Cond) {
	mutx.Lock()
	defer func() {
		log.Print(gid + " 执行defer()\n")
		mutx.Unlock()
	}()

    // 等待打破条件
	for !sig.ready {
	    // 等待ready信号
		log.Print(gid + " 等待唤醒...\n")
		cd.Wait()
	}

	log.Print(gid + " 被叫醒! \n")
}

func Condition(sig *signal, gid string, mutx *sync.Mutex, cd *sync.Cond) {
	mutx.Lock()
	defer func() {
		log.Print(gid + " 执行defer()\n")
		mutx.Unlock()
	}()

	log.Print(gid + " do something...\n")
	time.Sleep(3 * time.Second)
	sig.ready = true
	log.Print(gid + "旋转开关，唤醒等待方...\n")
	cd.Signal()
}

// 协程通信信号
type signal struct {
	// 用于wait()条件的检测
	ready bool
}
func main() {
    // 构建互斥共享变量
	mtx := new(sync.Mutex)
	cond := sync.NewCond(mtx)
	signal := new(signal)
	
	go ForWhileSignal(signal, "G-等待方", mtx, cond)
	time.Sleep(100)
	go Condition(signal, "G-唤醒方", mtx, cond)

	for {
		time.Sleep(1)
	}
}

type Pool struct {
	*Option
	idle    *list.List  // 空闲队列(双向)链表
	actives int         // 总连接数
	mtx     *sync.Mutex // 同步锁
	cond    *sync.Cond  // 用于阻塞/唤醒
}

// NewPool 创建连接池,初始化连接队列,更新队列大小
func NewPool(opt *Option) (p *Pool, err error) {
	idle := list.New()
	var conn *Conn
	for i := 0; i < opt.size; i++ {
		conn, err = NewConn(opt)
		if err == nil {
			// 加入队列
			idle.PushBack(conn)
		}
		// whether close all idle conn when one of err occurs?
	}

    // mutex用于隔离临界区
	mutx := new(sync.Mutex)
	// cond用于唤醒/等待当前函数块
	cond := sync.NewCond(mutx)

	p = &Pool{
		opt,
		idle,
		idle.Len(),
		mutx,
		cond,
	}
	return
}

/*
	Get 获取连接:
	空闲列表是否有库存:
		- 没有
			- 连接数是否达到上限
				- 是, 解锁, 阻塞等待唤醒
				- 否, 创建连接
		- 从队头出队获取
*/
func (p *Pool) Get() (c *Conn, err error) {
	p.mtx.Lock()
	defer p.mtx.Unlock()
	// 如果当前活跃大于限制数量, 阻塞等待
	for p.idle.Len() == 0 && p.actives >= p.size {
		log.Print("idle size full, blocking...")
		// 让出使用权并且释放mutex锁，当for条件不成立会自动唤醒
		p.cond.Wait()
	}
	// 空闲列表如果有且连接, 则从空闲队头获取
	if p.idle.Len() > 0 {
		c = p.idle.Remove(p.idle.Front()).(*Conn)
	} else {
		// 创建连接
		c, err = NewConn(p.Option)
		if err == nil {
			p.actives++
		}
	}
	return
}

/*
  Put() 归还连接
  - 连接是否异常
	- 是, 关闭异常连接
	- 否, 归还连接至队尾
  - 更新占用连接数, 唤醒等待方
*/
func (p *Pool) Put(c *Conn, err error) {
	p.mtx.Lock()
	defer p.mtx.Unlock()

	if err != nil {
	    // 如果当前连接异常，执行关闭
		if c != nil {
			c.Close()
		}
	} else {
	    // 健康连接放入队尾
		p.idle.PushBack(c)
	}
	p.actives--
	// 唤醒等待队列
	p.cond.Signal()
}

var opt = &Option{
	addr:        "127.0.0.1:3000",
	// 初始化5个连接
	size:        5,
	readTimeout: 30 * time.Second,
	dialTimeout: 5 * time.Second,
	keepAlive:   30 * time.Second,
}

func TestNewPool(t *testing.T) {
	pool, err := NewPool(opt)
	if err != nil {
		t.Fatal(err)
	}

	for i := 0; i < 10; i++ {
		go func(id int) {
			if err := SendInPool(pool, "Uid-"+strconv.Itoa(id)); err != nil {
				log.Print("Send in pool err: ", err)
			}
			// 注意闭包的id传入值
		}(i)
	}

    // just holding
	for {
		time.Sleep(1)
	}
}

func SendInPool(p *Pool, uid string) (err error) {
	var c *Conn
	if c, err = p.Get(); err != nil {
		return
	}
	defer p.Put(c, err)
	msg := &body.Message{Uid: uid, Val: "pixelpig!"}
	rec, err := c.Send(context.Background(), msg)
	if err != nil {
		log.Print(err)
	} else {
		log.Print(uid, ", Msg: ", <-rec)
	}
	return
}

=== RUN   TestNewPool
2021/05/17 01:22:59 idle size full, blocking...
2021/05/17 01:22:59 idle size full, blocking...
2021/05/17 01:22:59 idle size full, blocking...
2021/05/17 01:22:59 idle size full, blocking...
2021/05/17 01:22:59 idle size full, blocking...
2021/05/17 01:22:59 Uid-4, Msg: ts(ns): 1621185779673605400, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-0, Msg: ts(ns): 1621185779673605400, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-6, Msg: ts(ns): 1621185779673605400, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-9, Msg: ts(ns): 1621185779672609700, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-3, Msg: ts(ns): 1621185779673605400, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-1, Msg: ts(ns): 1621185779690594100, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-8, Msg: ts(ns): 1621185779690594100, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-5, Msg: ts(ns): 1621185779690594100, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-2, Msg: ts(ns): 1621185779690594100, server: hello, pixelpig!
2021/05/17 01:22:59 Uid-7, Msg: ts(ns): 1621185779690594100, server: hello, pixelpig!

type Pool struct {
	m        sync.Mutex
	//可以管理任意实现了Closer接口的资源类型
	resource chan io.Closer
	maxSize  int
	usedSize int
	// 构建连接的工厂函数，返回实现io.Closer的实例
	factory  func() (io.Closer, error)
	closed   bool
}

// 如果Conn没有实现io.Closer接口，则编译器会提示：
// Cannot use 'new(Conn)' (type *Conn) as type io.Closer 
// Type does not implement 'io.Closer'

var _ io.Closer = new(Conn)

//get resource
func (p *Pool) Acquire() (io.Closer, error) {
	p.m.Lock()
	defer p.m.Unlock()

	select {
	// 有闲置连接则取出
	case r, ok := <-p.resource:
		log.Println("Acquire:", "Shard Resource")
		if !ok {
			//管道已经关闭
			return nil, ErrPoolClosed
		}
		return r, nil
	default:
		if p.usedSize < p.maxSize {
			p.usedSize++
			log.Printf("Acquire:" + "New Resource." +
				"resource present size/max: %d/%d\n", p.usedSize, p.maxSize)
			return p.factory()
		} else {
			//log.Printf("Acquire:" +
			//	"block for pool's dry, present size: %d/%d", p.usedSize, p.maxSize)
			return nil, nil
		}
	}
}

func (p *Pool) Release(r io.Closer) {
	p.m.Lock()
	defer p.m.Unlock()

	if p.closed {
		r.Close()
		return
	}

	p.usedSize--

	select {
	//资源放回队列
	case p.resource <- r:
		log.Println("Release:", "into queue")
	//队列满的情况关闭资源
	default:
		log.Println("Release:", "Closing")
		r.Close()
	}
}