Compare commits

...

60 Commits

Author SHA1 Message Date
junleea 17b9bfc6f5 测试内存泄露问题 2025-03-10 15:50:24 +08:00
junleea df0985bc28 测试内存泄露问题 2025-03-10 15:34:05 +08:00
junleea 0cc6792be0 测试内存泄露问题 2025-03-10 15:25:58 +08:00
junleea 795cc35396 测试内存泄露问题 2025-03-10 15:22:09 +08:00
junleea b27a7fb7cc 测试内存泄露问题 2025-03-10 15:17:47 +08:00
junleea 8ceea2bb55 测试内存泄露问题 2025-03-10 15:15:46 +08:00
junleea fd4e0c0495 测试内存泄露问题 2025-03-10 15:13:30 +08:00
junleea 539ea9215b 测试内存泄露问题 2025-03-10 15:00:42 +08:00
junleea 8c382bbbff 测试内存泄露问题 2025-03-10 14:45:13 +08:00
junleea 6ab124b0cb 测试内存泄露问题 2025-03-10 14:39:15 +08:00
lijun ea963f4076 获取当前帧bytes,垃圾回收 2025-01-23 23:06:31 +08:00
lijun 6bdaeaf017 获取当前帧bytes,垃圾回收 2025-01-23 23:03:37 +08:00
lijun 2040232735 获取当前帧bytes 2025-01-23 22:50:45 +08:00
lijun 7c1d9559b9 获取当前帧bytes 2025-01-23 22:49:09 +08:00
lijun 5f2b5b55f4 获取当前帧bytes 2025-01-23 22:43:20 +08:00
lijun 2a120d85f4 获取当前帧bytes 2025-01-23 22:40:07 +08:00
lijun c6bc92c4de 获取当前帧bytes 2025-01-23 22:31:04 +08:00
lijun 9362523c39 内存泄漏解决测试 2025-01-22 15:09:15 +08:00
lijun fdc3407b6d 内存泄漏解决测试 2025-01-22 14:46:23 +08:00
lijun 02a7738c49 内存泄漏解决测试 2025-01-22 14:28:04 +08:00
lijun 1110ca1b9c 内存泄漏解决测试 2025-01-22 14:22:15 +08:00
lijun 7e7cc07c4d 内存泄漏解决测试 2025-01-22 14:15:35 +08:00
lijun bf6546069b 内存泄漏解决测试 2025-01-22 14:03:03 +08:00
lijun c0219ce83c 内存泄漏解决测试 2025-01-22 13:57:11 +08:00
lijun 4d7521e4b5 内存泄漏解决测试 2025-01-22 13:51:04 +08:00
lijun f1401e8a9b 使用v3获取当前帧 2025-01-20 18:56:42 +08:00
lijun 6486b8328b 使用v3获取当前帧 2025-01-20 18:52:50 +08:00
lijun 07cb9515be 使用v3获取当前帧 2025-01-20 18:49:45 +08:00
lijun e969fc53a2 使用v3获取当前帧 2025-01-20 18:30:33 +08:00
lijun 90e64920c7 使用v3获取当前帧 2025-01-20 18:14:33 +08:00
lijun b155ce2c1d 使用v3获取当前帧 2025-01-20 18:13:39 +08:00
lijun 7d55d68c90 接口查看设备是否在获取图像 2025-01-19 15:44:18 +08:00
lijun b00537b5ec release版本,输出每小时一次 2025-01-19 15:13:51 +08:00
lijun 2b3724a00a release版本,输出每小时一次 2025-01-18 21:14:21 +08:00
lijun 1e6a0996a0 release版本,输出每小时一次 2025-01-18 20:57:39 +08:00
lijun f86c1fd3a4 使用分离方式,修改视频流返回方式,使用redis存是否在获取 2025-01-18 20:50:33 +08:00
lijun 8827fa0412 使用分离方式,修改视频流返回方式,使用redis存是否在获取 2025-01-18 20:43:28 +08:00
lijun cead913af7 使用分离方式,修改视频流返回方式,使用redis存是否在获取 2025-01-18 20:38:35 +08:00
lijun 0906fce8cf 使用分离方式,修改视频流返回方式,使用redis存是否在获取 2025-01-18 20:24:51 +08:00
lijun 5552548a35 使用分离方式,修改视频流返回方式,使用redis存是否在获取 2025-01-18 20:21:38 +08:00
lijun de9b8d23ac 使用分离方式,修改视频流返回方式 2025-01-18 20:12:16 +08:00
lijun abea9bd374 使用分离方式,修改视频流返回方式 2025-01-18 20:04:50 +08:00
lijun d69e02115f 使用分离方式,修改视频流返回方式 2025-01-18 19:59:32 +08:00
lijun 69f81addeb 内存泄漏问题,注释日志输出 2025-01-18 16:08:07 +08:00
lijun 1d48452f23 内存泄漏问题,新拷贝 2025-01-18 16:04:54 +08:00
lijun 1e8b72e9b2 内存泄漏问题,深拷贝 2025-01-18 16:01:42 +08:00
lijun 5bd2707ffd 内存泄漏问题,深拷贝 2025-01-18 15:57:11 +08:00
lijun 74d04602d5 内存泄漏问题,深拷贝 2025-01-18 15:54:38 +08:00
lijun a202a2ef4c 内存泄漏问题,深拷贝 2025-01-18 15:48:20 +08:00
lijun 318907eeac 内存泄漏问题,深拷贝 2025-01-18 15:41:15 +08:00
lijun a965a9dd7d 内存泄漏问题,深拷贝 2025-01-18 15:38:10 +08:00
lijun 66e341e938 内存泄漏问题 2025-01-18 15:32:10 +08:00
lijun b8b6cb838c 内存泄漏问题 2025-01-18 15:26:04 +08:00
lijun 6aeb4dd4f8 内存泄漏问题 2025-01-18 15:17:20 +08:00
lijun 87c3522374 内存泄漏问题 2025-01-18 15:13:51 +08:00
lijun 59ae9d7f6f 内存泄漏问题 2025-01-18 15:10:15 +08:00
lijun 4c1a397775 内存泄漏问题 2025-01-18 14:55:02 +08:00
lijun 8fbc54c2af 内存泄漏问题 2025-01-18 14:38:33 +08:00
lijun 749eb5c2ee 使用分离的全局变量存储帧,不用map 2025-01-18 14:34:20 +08:00
lijun f83e720224 使用分离的全局变量存储帧,不用map 2025-01-18 14:21:04 +08:00
5 changed files with 226 additions and 207 deletions

3
go.mod
View File

@ -11,7 +11,8 @@ require (
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.3
github.com/robfig/cron/v3 v3.0.1
gocv.io/x/gocv v0.39.0
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6
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

2
go.sum
View File

@ -106,6 +106,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
github.com/ulule/deepcopier v0.0.0-20200430083143-45decc6639b6 h1:TtyC78WMafNW8QFfv3TeP3yWNDG+uxNkk9vOrnDu6JA=
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=

View File

@ -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"
@ -34,7 +33,7 @@ func SetUpToolGroup(router *gin.Engine) {
}
// 跨域访问cross origin resource share
func CrosHandler() gin.HandlerFunc {
func CorsHandler() gin.HandlerFunc {
return func(context *gin.Context) {
//method := context.Request.Method
context.Writer.Header().Set("Access-Control-Allow-Origin", "*")
@ -65,152 +64,171 @@ type videoStreamReq struct {
}
func GetVideoStream(c *gin.Context) {
var req videoStreamReq
id, _ := c.Get("id")
id1 := id.(int)
deviceIDSTR := c.Query("id")
deviceID, err := strconv.Atoi(deviceIDSTR)
if err != nil {
c.JSON(400, gin.H{"error": "device_id error"})
return
}
key := c.Query("key")
//校验权限
device := service.GetDevice(req.ID, id1)
device := service.GetDevice(deviceID, id1)
if device.ID == 0 {
c.JSON(400, gin.H{"error": "device not exist"})
return
}
if err := c.ShouldBind(&req); err != nil {
c.JSON(400, gin.H{"error": err.Error()})
return
} else {
//查看id是否存在
index := -1
for _, device := range proto.Config.DeviceInfo {
if device.ID == req.ID {
index = req.ID
break
}
}
if index == -1 {
c.JSON(400, gin.H{"error": "id not exist"})
return
}
//查看key是否正确
if req.Key != "123456" {
c.JSON(400, gin.H{"error": "key error"})
return
}
//设备流
c.Stream(func(w io.Writer) bool {
var count int
for {
frame, cnt := service.GetDeviceCurrentFrame(req.ID)
if cnt == count {
time.Sleep(50 * time.Millisecond)
continue
}
//gocv.Matrix转为jpeg
img, err := gocv.IMEncode(".jpg", frame)
frame_ := img.GetBytes()
_, err = w.Write([]byte("--frame\r\nContent-Type: image/jpeg\r\n\r\n"))
if err != nil {
fmt.Printf("写入头部信息错误: %v\n", err)
return false
}
_, err = w.Write(frame_)
if err != nil {
fmt.Printf("写入帧数据错误: %v\n", err)
return false
}
_, err = w.Write([]byte("\r\n"))
if err != nil {
fmt.Printf("写入帧结束标记错误: %v\n", err)
return false
}
time.Sleep(50 * time.Millisecond) // 控制帧率模拟每秒约20帧可按实际调整
}
})
rKey := worker.GetRedis("video_stream_get_stream_key")
if rKey == "" {
rKey = "123456"
}
//查看id是否存在
index := -1
for _, device_ := range proto.Config.DeviceInfo {
if device_.ID == deviceID {
index = deviceID
break
}
}
if index == -1 {
c.JSON(400, gin.H{"error": "id config not exist"})
return
}
//查看key是否正确
if key != rKey {
c.JSON(400, gin.H{"error": "key error"})
return
}
//查看设备是否在获取
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", deviceID))
if isGetting != "true" {
c.JSON(400, gin.H{"error": "device is not getting or not exist"})
log.Printf("stream device_id:%d is not getting or not exist", deviceID)
return
}
//设备流
c.Stream(func(w io.Writer) bool {
var count int
errCount := 0
for {
if errCount > 10 {
log.Printf("stream device:%d errCount > 10", deviceID)
return false
}
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)
errCount++
continue
}
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(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
}
})
}
// 发送实时视频流
func GetRealTimeImage(c *gin.Context) {
id, _ := c.Get("id")
id1 := id.(int)
device_id := c.Query("device_id")
device_id_int, _ := strconv.Atoi(device_id)
device := service.GetDevice(device_id_int, id1)
deviceId := c.Query("device_id")
deviceIdInt, _ := strconv.Atoi(deviceId)
device := service.GetDevice(deviceIdInt, id1)
if device.ID == 0 {
c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device not found"})
return
}
//查看设备是否在获取
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
if isGetting != "true" {
c.JSON(http.StatusOK, gin.H{"code": 4, "message": "device is not getting or not exist"})
log.Printf("device_id:%d is not getting or not exist", deviceIdInt)
return
}
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
log.Printf("connect wss err:%v", err)
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", device_id_int)
go subscribeAndHandleMessagesV3(ws, device_id_int)
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
}
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
tCount := 0
//计算帧率
for {
//从service获取当前帧
var img gocv.Mat
c := service.GetDeviceCurrentFrameV2(&img, device_id)
c := service.DeviceFrameCount[deviceId]
if c != count {
//将img转[]byte
if img.Empty() {
log.Printf("device:%d img is empty!", device_id)
} 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)
goto end
}
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)
goto end
}
c = count
err4 := img.Close()
if err4 != nil {
log.Printf("close img err:%v", err)
}
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
}
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)
worker.SetRedisSetRemove(online_conn_key, con_id)
goto end
break
} else {
log.Printf("Connection check success")
}
}
}
time.Sleep(100 * time.Millisecond)
t_count++
tCount++
}
end:
// 查看是否还有其他连接,没有则设置 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)
}
}

61
main.go
View File

@ -19,10 +19,11 @@ import (
)
var configPath string
var logCount int
func main() {
fmt.Println("start server")
gin.SetMode(gin.DebugMode) //设置为debug模式
gin.SetMode(gin.ReleaseMode) //设置为debug模式
r := gin.Default()
//数据库初始
err0 := dao.Init()
@ -35,7 +36,7 @@ func main() {
if err0 != nil {
panic("failed to connect redis:" + err0.Error())
}
r.Use(handler.CrosHandler())
r.Use(handler.CorsHandler())
r.Use(JWTAuthMiddleware()) // 使用 JWT 认证中间件
handler.SetUpToolGroup(r) // Tool
err := worker.InitRedis()
@ -51,6 +52,7 @@ func main() {
}
c.Start()
fmt.Println("定时任务已启动")
initDeviceGettingStatus()
err3 := r.Run(":" + proto.Config.SERVER_PORT)
if err3 != nil {
panic("failed to run server:" + err3.Error())
@ -66,18 +68,20 @@ func init() {
} else {
configPath = "/home/videoplayer/vp_stream.conf"
}
logCount = 0
//读取配置文件
err := proto.ReadConfig(configPath)
if err != nil {
panic("failed to read config file:" + err.Error())
}
service.DeviceFrameCount = make(map[int]int)
for _, device := range proto.Config.DeviceInfo {
//service.DeviceRWMap[device.ID] = sync.RWMutex{}
//service.DeviceCurrentFrameMap[device.ID] = gocv.NewMat()
//service.DeviceIsGettingFrame[device.ID] = false
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount.Store(device.ID, 0)
service.DeviceFrameCount[device.ID] = 0
service.DeviceIsGettingFrame.Store(device.ID, false)
}
}
@ -86,7 +90,20 @@ func myTask() {
ReadConfigAndSetSystem()
}
func initDeviceGettingStatus() {
for _, device := range proto.Config.DeviceInfo {
if device.NextStop == false {
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
}
if device.NextStop == true {
worker.DelRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
}
}
}
func ReadConfigAndSetSystem() {
logCount++
//configPath := "/home/videoplayer/vp_stream.conf"
//读取配置文件
err := proto.ReadConfig(configPath)
@ -95,20 +112,31 @@ func ReadConfigAndSetSystem() {
}
//检测是否需要获取设备流,如果需要则开启
for _, device := range proto.Config.DeviceInfo {
is_get, ok := service.DeviceIsGettingFrame.Load(device.ID)
if !ok {
//isGet, ok := service.DeviceIsGettingFrame.Load(device.ID)
//isGet_ := isGet.(bool)
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", device.ID))
if isGetting == "" {
//说明没有这个设备,需初始化添加
service.DeviceRWMap.Store(device.ID, &sync.RWMutex{})
service.DeviceCurrentFrameMap.Store(device.ID, gocv.NewMat())
service.DeviceFrameCount.Store(device.ID, 0)
service.DeviceFrameCount[device.ID] = 0
service.DeviceIsGettingFrame.Store(device.ID, false)
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", device.ID), "false")
}
if is_get == false && device.NextStop == false { //如果设备流已经停止且不暂停,则开启
if isGetting == "false" && device.NextStop == false { //如果设备流已经停止且不暂停,则开启
switch device.ID {
case 50:
service.Device50CurrentFrame = gocv.NewMat()
case 73:
service.Device73CurrentFrame = gocv.NewMat()
}
go service.GetVideoStream(device.ID)
log.Printf("device:%d has started!\n", device.ID)
}
}
log.Println("每10秒执行一次,当前设备:", proto.Config.DeviceInfo)
if logCount%3600 == 0 {
log.Printf("每%d秒执行一次,当前设备:%v", logCount*10, proto.Config.DeviceInfo)
}
}
func JWTAuthMiddleware() gin.HandlerFunc {
@ -153,13 +181,16 @@ func JWTAuthMiddleware() gin.HandlerFunc {
if worker.IsContainSet("super_permission_tokens", tokenString) {
s_id := c.Request.Header.Get("super_id")
if s_id == "" {
c.AbortWithStatus(200)
c.JSON(200, gin.H{
"message": "NOT_LOGIN",
"error": "super_id is empty",
"code": 3,
})
return
s_id = c.Query("super_id")
if s_id == "" {
c.AbortWithStatus(200)
c.JSON(200, gin.H{
"message": "NOT_LOGIN",
"error": "super_id is empty",
"code": 3,
})
return
}
}
id, _ := strconv.Atoi(s_id)
//查看s_id类型

View File

@ -16,120 +16,78 @@ import (
var DeviceRWMap = &sync.Map{}
var DeviceCurrentFrameMap = &sync.Map{}
var DeviceFrameCount = &sync.Map{}
var DeviceFrameCount map[int]int
var DeviceIsGettingFrame = &sync.Map{}
var Device1CurrentFrame []byte
var Device50CurrentFrame gocv.Mat
var Device73CurrentFrame gocv.Mat
func SetDeviceCurrentFrame(frame gocv.Mat, device_id int) error {
func SetDeviceCurrentFrame(frame gocv.Mat, deviceId int) error {
//获取读写锁
mutex_, ok := DeviceRWMap.Load(device_id)
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
return fmt.Errorf("设备:%s 读写锁不存在", device_id)
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
}
mutex := mutex_.(*sync.RWMutex)
mutex.Lock()
defer mutex.Unlock()
//获取前一帧,将前一帧释放
framePrev, ok := DeviceCurrentFrameMap.Load(device_id)
framePrev, ok := DeviceCurrentFrameMap.Load(deviceId)
if ok {
frame_, ok2 := framePrev.(gocv.Mat)
if ok2 {
err2 := frame_.Close()
if err2 != nil {
log.Printf("设备:%d, 错误: 无法关闭帧\n", device_id)
log.Printf("设备:%d, 错误: 无法关闭帧\n", deviceId)
}
}
}
//设置当前帧
DeviceCurrentFrameMap.Store(device_id, frame)
frame_count, ok := DeviceFrameCount.Load(device_id)
DeviceCurrentFrameMap.Store(deviceId, frame)
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
return fmt.Errorf("设备:%s 当前帧计数不存在", device_id)
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
}
frame_count_ := frame_count.(int)
frame_count_++
DeviceFrameCount.Store(device_id, frame_count_)
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)
}
}()
func SetDeviceCurrentFrameV2(frame *gocv.Mat, deviceId int) error {
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在device_id: %d \n", deviceId)
return -1
return fmt.Errorf("设备:%s 读写锁不存在", deviceId)
}
mutex, ok := mutex_.(*sync.RWMutex)
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型device_id: %d \n", deviceId)
return -1
}
mutex.RLock()
defer mutex.RUnlock()
var frame_ gocv.Mat
//获取当前帧
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
if !ok {
return -1
}
frame_, ok = frameIface.(gocv.Mat)
if !ok {
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型device_id: %d \n", deviceId)
}
*frame = frame_
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
if !ok {
return -1
}
frame_count, ok := frame_countIface.(int)
if !ok {
log.Printf("DeviceFrameCount 存储的不是 int 类型device_id: %d", deviceId)
}
return frame_count
}
func GetDeviceCurrentFrame(deviceId int) (gocv.Mat, int) {
defer func() {
if err := recover(); err != nil {
log.Printf("设备:%d 错误: %v\n", deviceId, err)
mutex := mutex_.(*sync.RWMutex)
mutex.Lock()
defer mutex.Unlock()
//设置当前帧
switch deviceId {
case 1:
Device1CurrentFrame = (*frame).ToBytes()
case 50:
if Device50CurrentFrame.Empty() {
Device50CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
}
}()
(*frame).CopyTo(&Device50CurrentFrame)
case 73:
if Device73CurrentFrame.Empty() {
Device73CurrentFrame = gocv.NewMatWithSize((*frame).Rows(), (*frame).Cols(), (*frame).Type())
}
(*frame).CopyTo(&Device73CurrentFrame)
//获取读写锁
mutex_, ok := DeviceRWMap.Load(deviceId)
if !ok {
log.Printf("DeviceRWMap 读写锁不存在device_id: %d \n", deviceId)
return gocv.NewMat(), -1
}
mutex, ok := mutex_.(*sync.RWMutex)
frameCount, ok := DeviceFrameCount[deviceId]
if !ok {
log.Printf("DeviceRWMap 存储的不是 *sync.RWMutex 类型device_id: %d \n", deviceId)
return gocv.NewMat(), -1
return fmt.Errorf("设备:%s 当前帧计数不存在", deviceId)
}
mutex.RLock()
defer mutex.RUnlock()
//获取当前帧
frameIface, ok := DeviceCurrentFrameMap.Load(deviceId)
if !ok {
return gocv.NewMat(), -1
if frameCount%10 == 0 {
log.Printf("设备:%d 当前帧: %d 已更新\n", deviceId, frameCount)
}
frame, ok := frameIface.(gocv.Mat)
if !ok {
log.Printf("DeviceCurrentFrameMap 存储的不是 gocv.Mat 类型device_id: %d \n", deviceId)
}
frame_countIface, ok := DeviceFrameCount.Load(deviceId)
if !ok {
return gocv.NewMat(), -1
}
frame_count, ok := frame_countIface.(int)
if !ok {
log.Printf("DeviceFrameCount 存储的不是 int 类型device_id: %d", deviceId)
}
return frame, frame_count
frameCount++
DeviceFrameCount[deviceId] = frameCount
return nil
}
func getVideoFrame(device proto.DeviceInfo) {
@ -157,6 +115,7 @@ func getVideoFrame(device proto.DeviceInfo) {
fontColor := color.RGBA{G: 255}
lineType := 2
z := 0
var frame gocv.Mat
for {
if device.LogFrame > 0 && z%device.LogFrame == 0 {
log.Printf("设备:%d 当前帧: %d\n", device.ID, z)
@ -164,7 +123,7 @@ func getVideoFrame(device proto.DeviceInfo) {
if device.NextStop {
break
}
frame := gocv.NewMat()
frame = gocv.NewMat()
ok := webcam.Read(&frame)
if !ok {
log.Printf("设备:%v 错误: 无法从视频流中读取帧\n", device)
@ -185,11 +144,21 @@ func getVideoFrame(device proto.DeviceInfo) {
currentTime := time.Now().Format("2006-01-02 15:04:05")
gocv.PutText(&frame, currentTime, image.Point{10, 20}, font, fontScale, fontColor, lineType)
//需要将帧付给全局变量
err3 := SetDeviceCurrentFrame(frame, device.ID)
err3 := SetDeviceCurrentFrameV2(&frame, device.ID)
if err3 != nil {
log.Printf("设备:%d 错误: 无法设置当前帧,err:%s \n", device.ID, err3.Error())
}
z++
//关闭帧
err4 := frame.Close()
if err4 != nil {
log.Printf("设备:%d 错误: 无法关闭帧,err:%s \n", device.ID, err4.Error())
}
}
//关闭帧
err4 := frame.Close()
if err4 != nil {
log.Printf("设备:%d 错误: 无法关闭帧,err:%s \n", device.ID, err4.Error())
}
}
@ -213,13 +182,9 @@ func GetVideoStream(id int) {
log.Printf("设备:%d 错误: %v\n", id, err)
}
}()
is_get, ok := DeviceIsGettingFrame.Load(id)
if !ok {
log.Printf("device: %d not found", id)
return
}
if is_get == true {
log.Printf("device: %d is running!", id)
isGetting := worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
if isGetting == "true" || isGetting == "" {
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
return
}
for {
@ -238,9 +203,9 @@ func GetVideoStream(id int) {
log.Printf("device: %d not found", id)
break
}
is_get, ok = DeviceIsGettingFrame.Load(id)
if is_get == true {
log.Printf("for device:%d is running!", id)
isGetting = worker.GetRedis(fmt.Sprintf("device_%d_is_getting", id))
if isGetting == "true" {
log.Printf("设备:%d 正在运行,isGetting:%s", id, isGetting)
break
}
@ -251,9 +216,11 @@ func GetVideoStream(id int) {
//设置设备控制信息
status := Get(device.Control)
DeviceIsGettingFrame.Store(id, true)
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "true")
log.Printf("device: %d set control info status: %d", device.ID, status)
getVideoFrame(device)
DeviceIsGettingFrame.Store(id, false)
worker.SetRedis(fmt.Sprintf("device_%d_is_getting", id), "false")
//等待1s
time.Sleep(1 * time.Second)
}