silenceper/pool

🚌 A golang general network connection poolction pool

pool

中文文档

A golang universal network connection pool.

Feature：

• More versatile, The connection type in the connection pool is interface{}, making it more versatile
• More configurable, The connection supports setting the maximum idle time, the timeout connection will be closed and discarded, which can avoid the problem of automatic connection failure when idle
• Support user setting ping method, used to check the connectivity of connection, invalid connection will be discarded
• Support connection waiting, When the connection pool is full, support for connection waiting (like the go db connection pool)

Basic Usage:

//factory Specify the method to create the connection
factory := func() (interface{}, error) { return net.Dial("tcp", "127.0.0.1:4000") }

//close Specify the method to close the connection
close := func(v interface{}) error { return v.(net.Conn).Close() }

//ping Specify the method to detect whether the connection is invalid
//ping := func(v interface{}) error { return nil }

//Create a connection pool: Initialize the number of connections to 5, the maximum idle connection is 20, and the maximum concurrent connection is 30
poolConfig := &pool.Config{
	InitialCap: 5,
	MaxIdle:   20,
	MaxCap:     30,
	Factory:    factory,
	Close:      close,
	//Ping:       ping,
	//The maximum idle time of the connection, the connection exceeding this time will be closed, which can avoid the problem of automatic failure when connecting to EOF when idle
	IdleTimeout: 15 * time.Second,
}
p, err := pool.NewChannelPool(poolConfig)
if err != nil {
	fmt.Println("err=", err)
}

//Get a connection from the connection pool
v, err := p.Get()

//do something
//conn=v.(net.Conn)

//Put the connection back into the connection pool, when the connection is no longer in use
p.Put(v)

//Release all connections in the connection pool, when resources need to be destroyed
p.Release()

//View the number of connections in the current connection pool
current := p.Len()

Remarks:

The connection pool implementation refers to pool https://github.com/fatih/pool , thanks.

License

The MIT License (MIT) - see LICENSE for more details

pool's Issues

请问下池子为什么每次要lock呢

请问下 pool 为什么每次要 lock ，是什么考虑呢

// getConns 获取所有连接
func (c *channelPool) getConns() chan *idleConn {
c.mu.Lock()
conns := c.conns
c.mu.Unlock()

示例代码错误

如题，创建一个池，现在应该是pool.Config{}，不是pool.PoolConfig{}，请调整下，，

关于.Unlock 提个问题，想了解一下实现

Put(conn interface{}) 这方法里

c.mu.Lock()后为什么不直接
defer c.mu.Unlock()

而是各个select 分支上再 c.mu.Unlock()
有什么特殊的考虑吗？

TestPool_Get测试无法通过

1 版本v1.0.0

TestPool_Get 测试方法中

	_, err = p.Get()
	if err != ErrMaxActiveConnReached {
		t.Errorf("Get error: %s", err)
	}

Get方法实现中

			c.mu.Lock()
			log.Debugf("openConn %v %v", c.openingConns, c.maxActive)
			if c.openingConns >= c.maxActive {
				req := make(chan connReq, 1)
				c.connReqs = append(c.connReqs, req)
				c.mu.Unlock()
				ret, ok := <-req
				if !ok {
					return nil, ErrMaxActiveConnReached
				}  			

ret, ok := <-req
目前在没有Put方法的情况下将永远阻塞 导致该测试无法完成

用户无法 Get 一个新创建的连接

背景：
上一次提 issue 的时候，我针对连接有效性提供了一个 ping 方法的口子，得到您的认同。
目的是解决 server 端重启，导致的连接失效的问题（连接池的所有连接都失效）。

然而，该问题在我这里还没有完全解决。

问题：
当我在使用 thrift client 中使用连接池的时候，server 端需要重启，导致连接池中的连接全部失效。
可是，像大部分 tcp 连接一样， thrift client 本身是没有办法检测这个连通性的，比如 ping 方法。

解决思路：

在查看 database/sql 包的时候，发现对于连接池的连接失效问题，处理的代码片段：

	for i := 0; i < maxBadConnRetries; i++ {
		res, err = db.exec(ctx, query, args, cachedOrNewConn)
		if err != driver.ErrBadConn {
			break
		}
	}
	if err == driver.ErrBadConn {
		return db.exec(ctx, query, args, alwaysNewConn)
	}

• 直接执行 exec 方法，从连接池获取一个连接 （cachedOrNewConn），如果报错为 driver.ErrBadConn，重试 maxBadConnRetries 次 （默认 2）
• 如果重试还是报错为 driver.ErrBadConn，创建新的 mysql 连接 （alwaysNewConn），来执行 exec 方法

从官方的处理方式中得到启示，我们也可以在 thrift rpc Call 调用报错的时候，重新创建连接，老连接直接丢弃。连接池中的失效连接会慢慢被丢弃干净。
类似，我们需要在 Get() 这个方法指定 alwaysNewConn，强制使用新创建的连接。

伪代码:

conn = pool.Get()
err = conn.exec("....")
if err == ErrBadConn{
        conn = pool.Get(alwaysNewConn)
        conn.exec("....")
}
pool.Put(conn)

其他思路：

• Release 释放连接池所有连接，重新 Get
  当 conn.exec("....") 报错的时候，通过 pool.Release() 释放所有连接的时候，然后 pool.Get() 获取的就是新创建的连接。

目前这个思路行不通，因为 Release 方法实际上是把整个 pool 关掉了，我们需要的是清空连接池中现有的连接。另外就是，过于粗暴，不能仅仅通过一个连接的连通性来断定其他连接的情况。

分配连接的时候有bug

mutex应该放到get里面，现在放在getConn里面多个线程可以同时拿到conn,所以同一个连接有几率会分配多次

factory 是否应该可置空？

如果用来存放客户端连接，没有连接的时候，总不能凭空造出来一个吧。

如果池子满了，客户端会直接返回error吗

我更希望的是，像go的db连接池那样，池子满了应该进入一个队列，等待一段时间，然后再返回错误。

文档有问题

&pool.PoolConfig

channelPool中的conns为什么要生命成指针类型

type channelPool struct {
mu sync.RWMutex
conns chan *idleConn
factory func() (interface{}, error)
close func(interface{}) error
ping func(interface{}) error
idleTimeout, waitTimeOut time.Duration
maxActive int
openingConns int
connReqs []chan connReq
}
里面的conns为啥要使用chan *idleConn，直接使用chan idleConn不可以吗？

Get 从连接池中获取的连接失效

假如连接池初始化后，连接的服务方关闭了，连接池中的所有连接就失效了。
因此，每次调用 Get 方法获取连接池中的一个连接时，需要类似 ping 的检查，失效的时候重新创建连接。

建议：
初始化的时候提供一个 Ping 方法，类似 factory close 的。

conns本身就是chan，没有必要再加读锁吧

pool/channel.go

Line 97 in 58e025e

conns := c.conns

连接超时关闭后，连接数为零，这个时候应该怎么办呢？

连接超时关闭后，连接数为零。这个时候是不是应该自动创建链接呢？

当超过最大链接数后程序进入阻塞状态

当超过最大链接数后程序进入阻塞状态，能否有报错机制，让程序不要进入阻塞

关于 `PUT` 方法的疑问

PUT 方法中的这个 req 是什么意思呢

当连接池中没有连接的时候，会一直去创建新的连接

MaxCap 这个只是限制了池中的连接的数量，并发量高的时候，会有大量的新的连接产生

grpc相关的周边包很多都停止更新3年了，发生了什么？

grpc相关的客户端连接池包好几个都停止更新3-5年了。
还有grpc接口参数验证器，很多也都停止更新了，
到底发生了什么？你们现在都在偏重哪个方向去开发了？

连接更新问题

pool/channel.go

Line 149 in 1f4530b

return c.Close(conn)

在Put方法中，如果发现连接池是满的，这时候是否可以把老的连接删除，把新的连接加进来？
这样可以实时更新连接池了吧
@silenceper @hopehook

return/continue?

1.case wrapConn := <-conns:
if wrapConn == nil {
return nil, ErrClosed
}
这种情况，应该是conn被close掉了吧，但是对象还在channel中，建议这里continue是不是更好呢？

全部连接被关闭后导致hang（阻塞）

现象：

当有较多的goroutine在等待从Pool中Get连接，同时没有空闲连接时，有种情况，当被使用的连接，同时出错并被调用者Close掉时，等待获取连接的goroutine会一直阻塞等待从connReq这个channel获取连接，而所有的连接都已经被关闭了，不会再有可用的连接，导致一直等待，程序hang住。

原因

在Get方法中，第133行检查完已创建的连接数达到了最大值后，会等待从connReq的channel等待其它goroutine Put回来的连接。
在Close方法中，会把openingConns--，但已经没法触发Get方法去新建连接了，Get方法都被阻塞在从connReq的channel拿连接了。

Get方法133行

pool/channel.go

Line 133 in 58e025e

if c.openingConns >= c.maxActive {

pool/channel.go

Lines 130 to 140 in 58e025e

	default:
	c.mu.Lock()
	log.Debugf("openConn %v %v", c.openingConns, c.maxActive)
	if c.openingConns >= c.maxActive {
	req := make(chan connReq, 1)
	c.connReqs = append(c.connReqs, req)
	c.mu.Unlock()
	ret, ok := <-req
	if !ok {
	return nil, ErrMaxActiveConnReached
	}

Close方法

pool/channel.go

Lines 202 to 213 in 58e025e

	func (c *channelPool) Close(conn interface{}) error {
	if conn == nil {
	return errors.New("connection is nil. rejecting")
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.close == nil {
	return nil
	}
	c.openingConns--
	return c.close(conn)
	}