线上 udp 客户端请求服务端客户端句柄泄漏问题

作者: 阿兵云原生 | 来源:发表于2023-09-04 23:31 被阅读0次

本题分别从如下三个方面来分享：

问题描述

自定义连接池的编写

common_pool 的使用

问题描述

线上有一个业务，某个通服务通知 udp 客户端通过向 udp 服务端（某个硬件设备）发送 udp 包来进行用户上线操作

当同时有大量的请求打到 udp 服务端的时候，udp 服务端的回包可能会在网络环境中丢包，（udp 是不可靠的）导致 udp 客户端不能及时的收到 udp 服务端的回包，在短时间内，udp 客户端的句柄又没有得到复用或者释放，没有收到回包的句柄就一直阻塞在那里，最终导致句柄泄漏

那么可以如何解决呢？

增大客户端的句柄数

使用连接池并且在读取服务端响应数据时加上超时时间

显然，第一个解决方式治标不治本，改大句柄数，当请求量变大的时候，仍然会出现句柄泄漏的情况

第二种方式相对靠谱很多

首先，咱们将发送 udp 包给服务端后，等待读取服务端的回包时，设置超时时间，超时后读取失败，释放或者归还句柄

维护一个内部的连接池，减少每一次创建句柄消耗的资源和时间，使用的时候从池子里面获取句柄，使用完毕之后再归还句柄

自定义连接池的编写 customer_pool

那么对于连接池，我们实际上是可以自己来进行造轮子的，仅用于学习，实际使用的话，自然还是会去使用经过大众考研过的公共开源库，我们可以来基本的分析和研究一下一个连接池需要有些什么？

创建池子，关闭池子，池子的关闭状态

从池子中获取连接，归还连接，销毁当前连接

池子中能容纳的最大连接数，最小连接数，当前连接数

根据当前实际的连接数来对池子进行扩容和缩容

池子中创建连接的函数具体实现

当然，我们自己来体会一下连接池以及演示上述 udp 的 demo，我们仅实现如下几个简单功能作为演示

创建池子，池子的关闭状态

从池子中获取连接，归还连接

池子中能容纳的最大连接数，最小连接数，当前连接数

池子中创建连接的函数具体实现

对于池子中具体链接的销毁，池子的关闭，池子的扩缩容，以及其他高级使用，xdm 可以进行扩展

customer_pool demo

自定义连接池，实际上咱们是使用 chan 通道来进行实现，具体源码可以查看：https://github.com/qingconglaixueit/customer_pool/blob/master/customer_pool/pool.go

定义连接池 MyConnPool 数据结构，和创建连接池
- MyConnPool 结构中的 sync.Mutex 主要是用于控制多协程中非 pool 成员的其他成员的互斥，我们知道 chan 内部是有锁进行控制的

[图片上传失败...(image-d8e09c-1693927520946)]

[图片上传失败...(image-e40e85-1693927520946)]

获取对象的具体实现

从池子中获取对象，如果获取不到则默认查看当前的池子状态是否可以创建新的连接

若可以，则直接创建连接，返回对象

此处在进行池子成员的变动时，需要加锁进行控制

func (conn *MyConnPool) GetObject() (interface{}, error) {
   return conn.getObject()
}
func (conn *MyConnPool) getObject() (interface{}, error) {
   if conn.isClosed {
      return nil, errors.New("pool is closed")
   }
   // 从通道里面读，如果通道里面没有则新建一个
   select {
   case object := <-conn.pool:
      return object, nil
   default:

   }
   // 校验当前的连接数是否大于最大连接数，若是，则还是需要从 pool 中取
   // 此时使用 mutex 主要是为了锁 MyConnPool 的非通道的其他成员
   conn.Lock()
   if conn.currentConn >= conn.maxConn {
      object := <-conn.pool
      conn.Unlock()
      return object, nil
   }
   // 逻辑走到此处需要新建对象放到 pool 中
   object, err := conn.connFun()
   if err != nil {
      conn.Unlock()
      return nil, fmt.Errorf("create conn error : %+v", err)
   }
   // 当前 pool 已有连接数+1
   conn.currentConn++
   conn.Unlock()

   return object, nil
}

释放对象的具体实现

使用完毕对象之后，需要归还

具体归还操作，则是将具体的连接再丢回通道里面即可

func (conn *MyConnPool) ReturnObject(object interface{}) error {
   return conn.returnObject(object)
}
func (conn *MyConnPool) returnObject(object interface{}) error {
   if conn.isClosed {
      return errors.New("pool is closed")
   }
   conn.pool <- object
   return nil
}

具体的应用

简单写一个 udp 服务端

可以查看源码地址：https://github.com/qingconglaixueit/use_common_pool/blob/master/server/main.go

代码注释部分用于测试超时的效果

[图片上传失败...(image-b7f155-1693927520946)]

使用咱们上述的自定义连接池编写客户端的 demo

具体源码地址：https://github.com/qingconglaixueit/customer_pool/blob/master/main.go

定义咱们有 udp 连接的对象

定义 PoolTest 对象，并简单的将 udp 连接加入到成员中

编写创建 udp 连接的函数 connectUdp

初始化连接池，设置池子最大 3 个连接，最小 1 个连接，实际创建连接函数为 connectUdp()

type PoolTest struct {
   Conn *net.UDPConn
}

var myPool *customer_pool.MyConnPool

func init() {
   myPool = customer_pool.NewMyConnPool(3, 1, func() (interface{}, error) {
      return connectUdp()
   })
   if myPool == nil {
      log.Panicln("NewMyConnPool error")
      return
   }
   log.Println("myPool == ", myPool)
}
// 创建连接函数
func connectUdp() (*PoolTest, error) {
   // 创建一个 udp 句柄
   log.Println(">>>>> 创建一个 udp 句柄 ... ")
   // 连接服务器
   conn, err := net.DialUDP("udp", nil, &net.UDPAddr{
      IP:   net.IPv4(127, 0, 0, 1),
      Port: 9998,
   })

   if err != nil {
      log.Println("Connect to udp server failed,err:", err)
      return nil, err
   }
   log.Printf("<<<<<< new udp connect %+v", conn)
   return &PoolTest{Conn: conn}, nil
}

获取到连接对象之后，咱们给 udp server 写入数据

GetObject() 获取具体的对象，获取到连接

SendMsg 进行具体消息的发送

ReturnObject() 将具体的对象归还到池子中

其中代码被注释掉的部分，是用力验证超时效果的，感兴趣的 xdm 可以将代码打开尝试一波

[图片上传失败...(image-2d4648-1693927520946)]

效果展示

最后补充上咱们的 main 函数，就可以进行测试验证了

func main() {
   for i := 0; i < 10; i++ {
      msg := fmt.Sprintf("send udp data is %d", i)
      go SendMsg(msg)
   }

   time.Sleep(2 * time.Second)
}

启动咱们的 udp 客户端，和 udp 服务端，我们可以查看到如下效果

客户端效果：

同时启了 10 个协程，每一个协程都会去池子里面拿连接对象，如果池子有现成的则直接使用，如果没有现成的，那么就新建一个连接， 如果当前池子已创建连接已经等于最大值，那么只能等着池子中有连接归还的时候再进行分配

总的来说，当前 demo 只会创建 3 个 udp 连接句柄

[图片上传失败...(image-f11e14-1693927520947)]

服务端效果：

可以看到服务端收到的 10 个请求，实际上只有 3 个句柄在多次请求

再一次印证了客户端实际上确实只创建了 3 次 udp 句柄

[图片上传失败...(image-e424b-1693927520947)]

上述是自定义简单连接池的基本 demo，关于 udp 超时处理的内容就不做演示，感兴趣的 xdm 可以下载源码来进行查看效果

https://github.com/qingconglaixueit/customer_pool

使用 go-commons-pool

当然，我们大致知道连接池基本是都有哪些组成部分，可以如何玩之后，我们来应用一个 golang 通用的连接池 go-commons-pool， 源码地址为：https://github.com/jolestar/go-commons-pool

[图片上传失败...(image-fea63-1693927520947)]

use_common_pool demo

应用 go-commons-pool 咱们的 demo 仅验证该库的通用和便捷，对于上述我们自定义的池子，咱们使用到的 udp 涉及到的代码，可以基本不用变动，直接使用 go-commons-pool 直接网上套即可

和咱们自定义池子不一样的地方

init 初始化池子的方法和配置不一样

SendMsg 方法，实现时使用的池子句柄不一样

当然，go-commons-pool 会好太多

实际 demo 为：

其余截图上未体现的 connectUdp()，(this *PoolTest) SendMsg(data []byte) ，和上述自定义池子实现方式完全一致

此处初始化池子配置，咱们也是一样传入具体池子最大的对象数，使用池子的默认配置，传入咱们创建连接的具体函数 connectUdp()

[图片上传失败...(image-70d1fa-1693927520947)]

对于到 main 函数和 SendMsg 函数，咱们的用法和写法基本完全一致

[图片上传失败...(image-78cca8-1693927520947)]

自然，效果也是一样的

当然对于 go-commons-pool 池子还有其他很多有意思的东西，感兴趣的可以来一起阅读以下他的源码，如下为当前池子的基本数据结构和创建池子的代码，咱们可以根据这个结构来追以下代码

代码目录如下：

[图片上传失败...(image-387560-1693927520947)]

./pool.go

基本数据结构

创建池子代码

[图片上传失败...(image-ead873-1693927520947)]

至此，本文结束

文中涉及到的源码地址：

customer_pool https://github.com/qingconglaixueit/customer_pool

use_common_pool https://github.com/qingconglaixueit/use_common_pool

go-commons-pool https://github.com/jolestar/go-commons-pool

感谢阅读，欢迎交流，点个赞，关注一波再走吧

欢迎点赞，关注，收藏

朋友们，你的支持和鼓励，是我坚持分享，提高质量的动力

[图片上传失败...(image-32c0cf-1693927520947)]

好了，本次就到这里

技术是开放的，我们的心态，更应是开放的。拥抱变化，向阳而生，努力向前行。

我是阿兵云原生，欢迎点赞关注收藏，下次见~

网友评论

本文标题：线上 udp 客户端请求服务端客户端句柄泄漏问题

本文链接：https://www.haomeiwen.com/subject/rngcvdtx.html

延伸阅读

深度阅读

您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

线上 udp 客户端请求服务端客户端句柄泄漏问题

问题描述

自定义连接池的编写 customer_pool

customer_pool demo

定义连接池 MyConnPool 数据结构，和创建连接池

获取对象的具体实现

释放对象的具体实现

具体的应用