手造一个Go的Mq连接池

type Pool interface {
	Get(ctx context.Context) (io.Closer, error)
	Put(ctx context.Context, c io.Closer, forceClose bool) error
	Close() error
}

func (p *Slice) Get(ctx context.Context) (io.Closer, error) {
	//锁定
	p.mu.Lock()

	if p.closed {
		p.mu.Unlock()
		return nil, ErrPoolClosed
	}
	//获取空闲时间，即你关闭了之后他还能活多久的时间
	idleTimeout := time.Duration(p.conf.IdleTimeout)
	// Prefer a free item, if possible.
	//空闲的携程
	numFree := len(p.freeItem)
	for numFree > 0 {
		i := p.freeItem[0]
		copy(p.freeItem, p.freeItem[1:])
		p.freeItem = p.freeItem[:numFree-1]
		p.mu.Unlock()
		//判断空闲线程是否已到期，到期就释放
		if i.expired(idleTimeout) {
			i.close()
			p.mu.Lock()
			p.release()
		} else {
			return i.c, nil
		}
		numFree = len(p.freeItem)
	}

	// Out of free items or we were asked not to use one. If we're not
	// allowed to open any more items, make a request and wait.
	//判断设置里边的活跃线程数是否大于零且当前活跃线程>=设置的活跃线程
	//从数组里取已经存在的
	if p.conf.Active > 0 && p.active >= p.conf.Active {
		// check WaitTimeout and return directly
		if p.conf.WaitTimeout == 0 && !p.conf.Wait {
			p.mu.Unlock()
			return nil, ErrPoolExhausted
		}
		// Make the item channel. It's buffered so that the
		// itemOpener doesn't block while waiting for the req to be read.
		req := make(chan item, 1)
		//获取下一个req的key
		reqKey := p.nextRequestKeyLocked()
		p.itemRequests[reqKey] = req
		wt := p.conf.WaitTimeout
		p.mu.Unlock()

		// reset context timeout
        //重置超时时间
		if wt > 0 {
			var cancel func()
			_, ctx, cancel = wt.Shrink(ctx)
			defer cancel()
		}
		// Timeout the item request with the context.
		select {
		case <-ctx.Done()://当收到上下文结束时就从数组中把这个map的key删掉
			// Remove the item request and ensure no value has been sent
			// on it after removing.
			p.mu.Lock()
            //把这个io从map中剔除
			delete(p.itemRequests, reqKey)
			p.mu.Unlock()
			return nil, ctx.Err()
		case ret, ok := <-req://如果这个io收到消息，就返回这个
			if !ok {
				return nil, ErrPoolClosed
			}
			//如果此时的时间>创建时间+空闲等待时间
			if ret.expired(idleTimeout) {
				ret.close()
				p.mu.Lock()
				p.release()
			} else {
				return ret.c, nil
			}
		}
	}

	//新建
	p.active++ // optimistically
	p.mu.Unlock()
	c, err := p.New(ctx)
	if err != nil {
		p.mu.Lock()
		p.release()
		p.mu.Unlock()
		return nil, err
	}
	return c, nil
}

func (p *Slice) Put(ctx context.Context, c io.Closer, forceClose bool) error {
	p.mu.Lock()
	defer p.mu.Unlock()

	//true，直接关闭释放
	if forceClose {
		p.release()
		return c.Close()
	}
	//
	added := p.putItemLocked(c)
	if !added {
		p.active--
		return c.Close()
	}
	return nil
}

func (p *Slice) putItemLocked(c io.Closer) bool {
	if p.closed {
		return false
	}
	//当前数组已满，不给放回
	if p.conf.Active > 0 && p.active > p.conf.Active {
		return false
	}
	i := item{
		c:         c,
		createdAt: nowFunc(),
	}
    //这一块没看太明白，道行还不够
	if l := len(p.itemRequests); l > 0 {
		var req chan item
		var reqKey uint64
		for reqKey, req = range p.itemRequests {
			break
		}
		delete(p.itemRequests, reqKey) // Remove from pending requests.
		req <- i
		return true
	} else if !p.closed && p.maxIdleItemsLocked() > len(p.freeItem) {//已关闭且空闲数组数目小于设置的数目
		p.freeItem = append(p.freeItem, &i)
		return true
	}
	return false
}

func (p *Slice) Close() error {
	p.mu.Lock()
	if p.closed {
		p.mu.Unlock()
		return nil
	}
	if p.cleanerCh != nil {
		close(p.cleanerCh)
	}
	var err error
	for _, i := range p.freeItem {
		i.close()
	}
	p.freeItem = nil
	p.closed = true
	for _, req := range p.itemRequests {
		close(req)
	}
	p.mu.Unlock()
    //关闭后的回调函数，可以自己设定
	p.stop()
	return err
}

func (d *Dao)NewProducerToTool(field *ProducerToPoolField) (*ProducerToPool,error) {
	//count := WaitProducerMapLen()
	//log.Info("uid is:%s,WaitProducerMapLen:%d",field.Uid,WaitProducerMapCount)

	var p ProducerToPool

	//打开ExitWait，如果它关闭了的话
	if !exitWaitIsOpen(field.Uid) {
		openExitWait(field.Uid)
	}

	p = ProducerToPool{
		done:   make(chan struct{}),
		isConnected:false,
		field:field,
		onlyloopchan: make(chan struct{}),
		isSend: make(chan struct{}),
	}
	go p.handleReconnect()
	return &p,nil
}

//断联监听，一旦断联就不停的尝试重连
func (p *ProducerToPool)handleReconnect(){
	count := 0
	for{
		p.isConnected = false
		log.Info("uid is:%s,Attempting to connect",p.field.Uid)

		for  {
			i := p.connect()
			if i == 2 {

				//WaitProducerMapCountLock.Lock()
				//WaitProducerMapCount++
				//WaitProducerMapCountLock.Unlock()

				atomic.AddInt32(&WaitProducerMapCount, 1)

				WaitChan <- p

				select {
				case <-ExitWait:
					continue
				case <-p.onlyloopchan:
					continue
				case <- p.done:
					return
				}
			} else if i != 0 {
				count++
				log.Info("uid is:%s,reconnect:%d",p.field.Uid,count)
				//time.Sleep(reconnectDelay)
				continue
			}
			break
		}
		select {
		case <-p.done:
			return
		case <-p.notifyClose:
			return
		}
	}
}

//0:success 1:fail->reconnect 2:con fail
func (p *ProducerToPool)connect() (int){
	ctx, cancel := context.WithCancel(context.Background())
	io_conn, err := cleanrbmqpool.Get(ctx)
	if err != nil {
		log.Info("uid is:%s,get io to pool err:%v",p.field.Uid,err)
		//如果是连接不通，退出重连循环，开始等待
		cancel()
		return 2
	}

	if conn, ok := io_conn.(*amqp.Connection);ok {

		//判断拿到的链接是否已被关闭
		if conn.IsClosed() {
			//直接释放掉
			cleanrbmqpool.Put(context.Background(),conn,true)
			return 1
		}

		ch, err := conn.Channel()

		if err != nil {
			log.Info("uid is:%s,get channel err:%v",p.field.Uid,err)
			return 1
		}

		err = ch.ExchangeDeclare(
			p.field.Exchange,     // name
			p.field.ExchangeType, // type
			p.field.Durable,      // durable
			false, // auto-deleted
			false, // internal
			false, // noWait
			nil,   // arguments
		)

		if err != nil {
			log.Info("uid is:%s, channel exchange declare failed",p.field.Uid)
			return 1
		}
		ch.Confirm(false)
		p.changeConnection(conn, ch)
		p.isConnected = true
		p.isSend <- struct{}{}
		//关闭ExitWait，如果它打开的话
		//做一个通知
		if exitWaitIsOpen(p.field.Uid) {
			//唤醒之前先做一个判断，判断map长度是否非空
			if !WaitProducerMapIsEmpty(p.field.Uid) {
				log.Info("uid is:%s,wake up",p.field.Uid)

				//notifyCond.Broadcast()

				close(ExitWait)

				//清空map
				ClearWaitProducerMap()
				//置0
				WaitProducerMapCount = 0

			}


		}

		return 0
	} else {
		log.Info("uid is:%s,io to amqp.Connection fail",p.field.Uid)
		return 1
	}
}

func(d *Dao)selectWaitProducerMap(){
	key := ""
	for {
		select {
		case p := <- WaitChan:
			log.Info("uid is:%s,WaitProducerMapCount:%d",p.field.Uid, WaitProducerMapCount)

			if WaitProducerMapCount == 1 {
				log.Info("uid is:%s,only loop",p.field.Uid)
				key = p.field.Uid
				time.Sleep(5 * time.Second)
				p.onlyloopchan <- struct{}{}
			} else if WaitProducerMapCount > 1{
				//判断是否那个唯一循环
				if key == p.field.Uid {
					log.Info("uid is:%s,only loop",p.field.Uid)

					//WaitProducerMapCountLock.Lock()
					//WaitProducerMapCount--
					//WaitProducerMapCountLock.Unlock()
					atomic.AddInt32(&WaitProducerMapCount, -1)

					time.Sleep(5 * time.Second)
					p.onlyloopchan <- struct{}{}
					continue
				}
				//判断是否已经存在于map中


				log.Info("uid is:%s,WaitProducerMapCount <= MAX_WAIT_PRODUCER",p.field.Uid)
				//WaitProducerMapCountLock.Lock()
				if WaitProducerMapCount <= MAX_CACHE_PRODUCER {
					log.Info("uid is:%s,field add cache",p.field.Uid)
					//CacheProducers = append(CacheProducers, p.field)

					//进map
					WaitProducerMap.Store(p.field.Uid,p)

				} else {
					log.Info("uid is:%s,cache filled",p.field.Uid)
					close(p.done)
				}
				//WaitProducerMapCountLock.Unlock()
			} else {
				log.Info("uid is:%s,wait",p.field.Uid)
			}
		}
	}
}