From df0985bc28e80d6a32db8fe461d71ebcfeafc26d Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Mon, 10 Mar 2025 15:34:05 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=86=85=E5=AD=98=E6=B3=84?= =?UTF-8?q?=E9=9C=B2=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- handler/tool.go | 107 +------------------- service/tool.go | 261 +----------------------------------------------- 2 files changed, 5 insertions(+), 363 deletions(-) diff --git a/handler/tool.go b/handler/tool.go index 252e395..35d1cbc 100644 --- a/handler/tool.go +++ b/handler/tool.go @@ -11,7 +11,6 @@ import ( "io" "log" "net/http" - "runtime" "strconv" "time" ) @@ -119,7 +118,8 @@ func GetVideoStream(c *gin.Context) { log.Printf("stream device:%d errCount > 10", deviceID) return false } - buf, cnt := service.GetDeviceCurrentFrameV4(deviceID, count) + buf := make([]byte, 1024*1024) + cnt := -1 if cnt == count || cnt == -1 { time.Sleep(50 * time.Millisecond) 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] if c != count { count = c - buf := service.Device1CurrentFrame.ToBytes() - 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 + buf := service.Device1CurrentFrame err2 := ws.WriteMessage(websocket.BinaryMessage, buf) if err2 != nil { 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) } } - -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() -} diff --git a/service/tool.go b/service/tool.go index c4ce062..848bb11 100644 --- a/service/tool.go +++ b/service/tool.go @@ -18,7 +18,7 @@ var DeviceRWMap = &sync.Map{} var DeviceCurrentFrameMap = &sync.Map{} var DeviceFrameCount map[int]int var DeviceIsGettingFrame = &sync.Map{} -var Device1CurrentFrame gocv.Mat +var Device1CurrentFrame []byte var Device50CurrentFrame gocv.Mat var Device73CurrentFrame gocv.Mat @@ -65,10 +65,7 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error { //设置当前帧 switch deviceId { case 1: - if Device1CurrentFrame.Empty() { - Device1CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type()) - } - (*frame).CopyTo(&Device1CurrentFrame) + Device1CurrentFrame = (*frame).ToBytes() case 50: if Device50CurrentFrame.Empty() { Device50CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type()) @@ -92,260 +89,6 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error { DeviceFrameCount[deviceId] = frameCount 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) { //捕获异常