测试内存泄露问题

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6
-	gocv.io/x/gocv v0.39.0
+	gocv.io/x/gocv v0.40.0
 	gorm.io/driver/mysql v1.5.7
 	gorm.io/driver/postgres v1.5.11
 	gorm.io/gorm v1.25.12

go.sum

@@ -110,6 +110,8 @@ github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6 h1:TtyC78WMafNW8Q
 github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6/go.mod h1:h8272+G2omSmi30fBXiZDMkmHuOgonplfKIKjQWzlfs=
 gocv.io/x/gocv v0.39.0 h1:vWHupDE22LebZW6id2mVeT767j1YS8WqGt+ZiV7XJXE=
 gocv.io/x/gocv v0.39.0/go.mod h1:zYdWMj29WAEznM3Y8NsU3A0TRq/wR/cy75jeUypThqU=
+gocv.io/x/gocv v0.40.0 h1:kGBu/UVj+dO6A9dhQmGOnCICSL7ke7b5YtX3R3azdXI=
+gocv.io/x/gocv v0.40.0/go.mod h1:zYdWMj29WAEznM3Y8NsU3A0TRq/wR/cy75jeUypThqU=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
 golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc=
 golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=

handler/tool.go

@@ -8,7 +8,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
-	"gocv.io/x/gocv"
 	"io"
 	"log"
 	"net/http"
@@ -113,57 +112,41 @@ func GetVideoStream(c *gin.Context) {
 	//设备流
 	c.Stream(func(w io.Writer) bool {
 		var count int
-		frame := gocv.NewMat()
-		defer func() {
-			//关闭帧
-			err4 := frame.Close()
-			if err4 != nil {
-				log.Printf("device:%d frame close err:%v", deviceID, err4)
-			}
-		}()
 		errCount := 0
 		for {
 			if errCount > 10 {
 				log.Printf("stream device:%d errCount > 10", deviceID)
 				return false
 			}
-			cnt := service.GetDeviceCurrentFrameV2(&frame, deviceID)
+			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)
+				log.Printf("stream device:%d,cnt =%d,count=%d,errCount=%d", deviceID, cnt, count, errCount)
 				errCount++
 				continue
 			}
-			if frame.Empty() {
-				log.Printf("stream device:%d frame is empty", deviceID)
-				time.Sleep(100 * time.Millisecond)
-				errCount++
-				continue
-			}
-			//gocv.Matrix转为jpeg
-			img, err2 := gocv.IMEncode(".jpg", frame)
-			if err2 != nil {
-				log.Printf("stream img encode err:%v", err2)
-				return false
-			}
-			frame_ := img.GetBytes()
-
+			count = cnt
 			_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
 			if err != nil {
 				fmt.Printf("写入头部信息错误: %v\n", err)
+				buf = nil
 				return false
 			}
-			_, err = w.Write(frame_)
+			_, err = w.Write(buf)
 			if err != nil {
 				fmt.Printf("写入帧数据错误: %v\n", err)
+				buf = nil
 				return false
 			}
 			_, err = w.Write([]byte("\r\n"))
 			if err != nil {
 				fmt.Printf("写入帧结束标记错误: %v\n", err)
+				buf = nil
 				return false
 			}
 			time.Sleep(50 * time.Millisecond) // 控制帧率，模拟每秒约20帧，可按实际调整
+			buf = nil
 		}
 	})
 }
@@ -193,60 +176,42 @@ func GetRealTimeImage(c *gin.Context) {
 		return
 	}
 	worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5)
-	log.Printf("device_id:%d  has set is_play to 1", deviceIdInt)
 	isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
 	if isGetting != "true" {
 		c.JSON(http.StatusOK, gin.H{"code": 5, "message": "device is not getting or not exist"})
 		log.Printf("device_id:%d is not getting or not exist", deviceIdInt)
 		return
 	}
-	go subscribeAndHandleMessagesV3(ws, deviceIdInt)
+	subscribeAndHandleMessagesV6(ws, deviceIdInt)
 }
 
-func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
+func subscribeAndHandleMessagesV6(ws *websocket.Conn, deviceId int) {
 	// 生成唯一连接 uuid
 	con_id := uuid.New().String()
-	online_conn_key := "device_" + strconv.Itoa(device_id) + "_online_conn_ids"
+	online_conn_key := "device_" + strconv.Itoa(deviceId) + "_online_conn_ids"
 	// 加入设备在线连接集合
 	worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
 	//图片计数器
 	count := 0
 	//定时器，发送计数器
-	t_count := 0
-	img := gocv.NewMat()
+	tCount := 0
+	//计算帧率
 	for {
 		//从service获取当前帧
-		c := service.GetDeviceCurrentFrameV2(&img, device_id)
+		c := service.DeviceFrameCount[deviceId]
 		if c != count {
-			if c == -1 {
-				log.Printf("device:%d get frame err!", device_id)
+			count = c
+			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
 			}
-			//将img转[]byte
-			if img.Empty() {
-				log.Printf("device:%d img is empty! count = %d \n", device_id, c)
-			} else {
-				//gocv.Matrix转为jpeg
-				buf, err := gocv.IMEncode(".jpg", img)
-				if err != nil {
-					log.Printf("img encode err:%v", err)
-					worker.SetRedisSetRemove(online_conn_key, con_id)
-					break
-				}
-				buf1 := buf.GetBytes()
-
-				err2 := ws.WriteMessage(websocket.BinaryMessage, buf1)
-				if err2 != nil {
-					log.Printf("send message to client err:%v", err2)
-					worker.SetRedisSetRemove(online_conn_key, con_id)
-					break
-				}
-				c = count
-			}
+			c = count
 		} else {
 			//每秒发送一次心跳检测
-			if t_count%10 == 0 {
+			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)
@@ -258,17 +223,12 @@ func subscribeAndHandleMessagesV3(ws *websocket.Conn, device_id int) {
 			}
 		}
 		time.Sleep(100 * time.Millisecond)
-		t_count++
-	}
-	// 关闭img
-	err := img.Close()
-	if err != nil {
-		log.Printf("device:%d img close err:%v", device_id, err)
+		tCount++
 	}
 
 	// 查看是否还有其他连接，没有则设置 is_play 为 0
 	if worker.IsContainKey(online_conn_key) == false {
-		worker.SetRedisWithExpire(strconv.Itoa(device_id)+"_is_play", "1", time.Minute*5)
-		log.Printf("device_id: %d has set is_play to 0", device_id)
+		worker.SetRedisWithExpire(strconv.Itoa(deviceId)+"_is_play", "1", time.Minute*5)
+		log.Printf("device_id: %d has set is_play to 0", deviceId)
 	}
 }

main.go

@@ -125,8 +125,6 @@ func ReadConfigAndSetSystem() {
 		}
 		if isGetting == "false" && device.NextStop == false { //如果设备流已经停止且不暂停，则开启
 			switch device.ID {
-			case 1:
-				service.Device1CurrentFrame = gocv.NewMat()
 			case 50:
 				service.Device50CurrentFrame = gocv.NewMat()
 			case 73:

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())
@@ -85,149 +82,14 @@ func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
 	if !ok {
 		return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
 	}
+	if frameCount%10 == 0 {
+		log.Printf("设备:%d 当前帧: %d 已更新\n", deviceId, frameCount)
+	}
 	frameCount++
 	DeviceFrameCount[deviceId] = frameCount
 	return nil
 }
 
-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) {
 	//捕获异常
 	defer func() {