ljydlv
/
VideoStream
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

测试内存泄露问题

This commit is contained in:
junleea 2025-03-10 15:34:05 +08:00
parent 0cc6792be0
commit df0985bc28
2 changed files with 5 additions and 363 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
handler/tool.go
View File

@ -11,7 +11,6 @@ import (
"io" "io"
"log" "log"
"net/http" "net/http"
"runtime"
"strconv" "strconv"
"time" "time"
) )
@ -119,7 +118,8 @@ func GetVideoStream(c *gin.Context) {
log.Printf("stream device:%d errCount > 10", deviceID) log.Printf("stream device:%d errCount > 10", deviceID)
return false return false
} }
buf, cnt := service.GetDeviceCurrentFrameV4(deviceID, count) buf := make([]byte, 1024*1024)
cnt := -1
if cnt == count || cnt == -1 { if cnt == count || cnt == -1 {
time.Sleep(50 * time.Millisecond) time.Sleep(50 * time.Millisecond)
log.Printf("stream device:%d,cnt =%d,count=%d,errCount=%d", deviceID, cnt, count, errCount) log.Printf("stream device:%d,cnt =%d,count=%d,errCount=%d", deviceID, cnt, count, errCount)
@ -201,56 +201,7 @@ func subscribeAndHandleMessagesV6(ws *websocket.Conn, deviceId int) {
c := service.DeviceFrameCount[deviceId] c := service.DeviceFrameCount[deviceId]
if c != count { if c != count {
count = c count = c
buf := service.Device1CurrentFrame.ToBytes() buf := service.Device1CurrentFrame
err2 := ws.WriteMessage(websocket.BinaryMessage, buf)
if err2 != nil {
log.Printf("send message to client err:%v", err2)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
}
buf = nil
c = count
} else {
//每秒发送一次心跳检测
if tCount%10 == 0 {
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
if err != nil {
log.Printf("Connection check failed:%v", err)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
} else {
log.Printf("Connection check success")
}
}
}
time.Sleep(100 * time.Millisecond)
tCount++
}
// 查看是否还有其他连接，没有则设置 is_play 为 0
if worker.IsContainKey(online_conn_key) == false {
worker.SetRedisWithExpire(strconv.Itoa(deviceId)+"_is_play", "1", time.Minute*5)
log.Printf("device_id: %d has set is_play to 0", deviceId)
}
}
func subscribeAndHandleMessagesV5(ws *websocket.Conn, deviceId int) {
// 生成唯一连接 uuid
con_id := uuid.New().String()
online_conn_key := "device_" + strconv.Itoa(deviceId) + "_online_conn_ids"
// 加入设备在线连接集合
worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
//图片计数器
count := 0
//定时器，发送计数器
tCount := 0
buf := make([]byte, 1024)
//计算帧率
for {
//从service获取当前帧
c := service.GetDeviceCurrentFrameV5(deviceId, count, buf)
if c != count {
count = c
err2 := ws.WriteMessage(websocket.BinaryMessage, buf) err2 := ws.WriteMessage(websocket.BinaryMessage, buf)
if err2 != nil { if err2 != nil {
log.Printf("send message to client err:%v", err2) log.Printf("send message to client err:%v", err2)
@ -281,55 +232,3 @@ func subscribeAndHandleMessagesV5(ws *websocket.Conn, deviceId int) {
log.Printf("device_id: %d has set is_play to 0", deviceId) log.Printf("device_id: %d has set is_play to 0", deviceId)
} }
} }
func subscribeAndHandleMessagesV3(ws *websocket.Conn, deviceId int) {
// 生成唯一连接 uuid
con_id := uuid.New().String()
online_conn_key := "device_" + strconv.Itoa(deviceId) + "_online_conn_ids"
// 加入设备在线连接集合
worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
//图片计数器
count := 0
//定时器，发送计数器
tCount := 0
var buf []byte
//计算帧率
for {
//从service获取当前帧
newBuf, c := service.GetDeviceCurrentFrameV4(deviceId, count)
if c != count {
buf = newBuf
count = c
err2 := ws.WriteMessage(websocket.BinaryMessage, buf)
if err2 != nil {
log.Printf("send message to client err:%v", err2)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
}
c = count
} else {
//每秒发送一次心跳检测
if tCount%10 == 0 {
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
if err != nil {
log.Printf("Connection check failed:%v", err)
worker.SetRedisSetRemove(online_conn_key, con_id)
break
} else {
log.Printf("Connection check success")
}
}
}
time.Sleep(100 * time.Millisecond)
tCount++
}
// 查看是否还有其他连接，没有则设置 is_play 为 0
if worker.IsContainKey(online_conn_key) == false {
worker.SetRedisWithExpire(strconv.Itoa(deviceId)+"_is_play", "1", time.Minute*5)
log.Printf("device_id: %d has set is_play to 0", deviceId)
}
//垃圾回收
runtime.GC()
}

261
service/tool.go
View File

@ -18,7 +18,7 @@ var DeviceRWMap = &sync.Map{}
var DeviceCurrentFrameMap = &sync.Map{} var DeviceCurrentFrameMap = &sync.Map{}
var DeviceFrameCount map[int]int var DeviceFrameCount map[int]int
var DeviceIsGettingFrame = &sync.Map{} var DeviceIsGettingFrame = &sync.Map{}
var Device1CurrentFrame gocv.Mat var Device1CurrentFrame []byte
var Device50CurrentFrame gocv.Mat var Device50CurrentFrame gocv.Mat
var Device73CurrentFrame gocv.Mat var Device73CurrentFrame gocv.Mat
@ -65,10 +65,7 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
//设置当前帧 //设置当前帧
switch deviceId { switch deviceId {
case 1: case 1:
if Device1CurrentFrame.Empty() { Device1CurrentFrame = (*frame).ToBytes()
Device1CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
}
(*frame).CopyTo(&Device1CurrentFrame)
case 50: case 50:
if Device50CurrentFrame.Empty() { if Device50CurrentFrame.Empty() {
Device50CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type()) Device50CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
@ -92,260 +89,6 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
DeviceFrameCount[deviceId] = frameCount DeviceFrameCount[deviceId] = frameCount
return nil return nil
} }
func GetDeviceCurrentFrameV5(deviceId int, curCount int, buf []byte) (cnt int) {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
// 异常时释放 buf
buf = nil
}
}()
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
return -1
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
return -1
}
mutex.RLock()
defer mutex.RUnlock()
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return -1
}
if frameCount == curCount {
return frameCount
}
//获取当前帧
var tempBuf []byte
switch deviceId {
case 1:
if Device1CurrentFrame.Empty() {
return -1
}
tempBuf = Device1CurrentFrame.ToBytes()
case 50:
if Device50CurrentFrame.Empty() {
return -1
}
tempBuf = Device50CurrentFrame.ToBytes()
case 73:
if Device73CurrentFrame.Empty() {
return -1
}
tempBuf = Device73CurrentFrame.ToBytes()
default:
return -1
}
// 避免直接返回大数组，可考虑复制必要的数据
if len(tempBuf) > 0 {
copy(buf, tempBuf)
// 释放临时变量
tempBuf = nil
}
return frameCount
}
func GetDeviceCurrentFrameV4(deviceId int, curCount int) (buf []byte, cnt int) {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
// 异常时释放 buf
buf = nil
}
}()
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
return nil, -1
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
return nil, -1
}
mutex.RLock()
defer mutex.RUnlock()
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return buf, -1
}
if frameCount == curCount {
return nil, frameCount
}
//获取当前帧
var tempBuf []byte
switch deviceId {
case 1:
if Device1CurrentFrame.Empty() {
return nil, -1
}
tempBuf = Device1CurrentFrame.ToBytes()
case 50:
if Device50CurrentFrame.Empty() {
return nil, -1
}
tempBuf = Device50CurrentFrame.ToBytes()
case 73:
if Device73CurrentFrame.Empty() {
return nil, -1
}
tempBuf = Device73CurrentFrame.ToBytes()
default:
return nil, -1
}
// 避免直接返回大数组，可考虑复制必要的数据
if len(tempBuf) > 0 {
buf = make([]byte, len(tempBuf))
copy(buf, tempBuf)
// 释放临时变量
tempBuf = nil
}
return buf, frameCount
}
func GetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) int {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
}
}()
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
return -1
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
return -1
}
mutex.RLock()
defer mutex.RUnlock()
//获取当前帧
switch deviceId {
case 1:
if (*frame).Empty() {
*frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
}
if Device1CurrentFrame.Empty() {
return -1
}
Device1CurrentFrame.CopyTo(frame)
case 50:
if (*frame).Empty() {
*frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
}
if Device50CurrentFrame.Empty() {
return -1
}
Device50CurrentFrame.CopyTo(frame)
case 73:
if (*frame).Empty() {
*frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
}
if Device73CurrentFrame.Empty() {
return -1
}
Device73CurrentFrame.CopyTo(frame)
}
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return -1
}
return frameCount
}
func GetDeviceCurrentFrameV3(deviceId int) (gocv.Mat, int) {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
}
}()
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
return gocv.NewMat(), -1
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
return gocv.NewMat(), -1
}
mutex.RLock()
defer mutex.RUnlock()
//获取当前帧
var frame gocv.Mat
switch deviceId {
case 1:
frame = gocv.NewMatWithSize(Device1CurrentFrame.Rows(), Device1CurrentFrame.Cols(), Device1CurrentFrame.Type())
Device1CurrentFrame.CopyTo(&frame)
//查看帧状态
//log.Printf("frame:%v,Device1CurrentFrame:%v \n", frame.Empty(), Device1CurrentFrame.Empty())
case 50:
frame = gocv.NewMatWithSize(Device50CurrentFrame.Rows(), Device50CurrentFrame.Cols(), Device50CurrentFrame.Type())
Device50CurrentFrame.CopyTo(&frame)
//查看帧状态
//log.Printf("frame:%v,Device50CurrentFrame:%v\n", frame.Empty(), Device50CurrentFrame.Empty())
case 73:
frame = gocv.NewMatWithSize(Device73CurrentFrame.Rows(), Device73CurrentFrame.Cols(), Device73CurrentFrame.Type())
Device73CurrentFrame.CopyTo(&frame)
}
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return gocv.NewMat(), -1
}
//查看地址
//log.Printf("frame:%p,Device1CurrentFrame:%p,Device50CurrentFrame:%p\n", &frame, &Device1CurrentFrame, &Device50CurrentFrame)
return frame, frameCount
}
func GetDeviceCurrentFrame(deviceId int) (gocv.Mat, int) {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
}
}()
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在，device_id: %d \n", deviceId)
return gocv.NewMat(), -1
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型，device_id: %d \n", deviceId)
return gocv.NewMat(), -1
}
mutex.RLock()
defer mutex.RUnlock()
//获取当前帧
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
if !ok {
return gocv.NewMat(), -1
}
frame, ok := frameIface.(gocv.Mat)
if !ok {
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型，device_id: %d \n", deviceId)
}
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return gocv.NewMat(), -1
}
return frame, frameCount
}
func getVideoFrame(device proto.DeviceInfo) { func getVideoFrame(device proto.DeviceInfo) {
//捕获异常 //捕获异常