From 357ada84550a73991d45ac0d6149b2a842b6a542 Mon Sep 17 00:00:00 2001 From: denghui <1016848185@qq.com> Date: Sat, 5 Jul 2025 10:59:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=20pc=20=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E7=AB=AF=E6=95=B0=E6=8D=AE=E4=B8=8A=E4=BC=A0=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/consts/emqx.go | 15 + .../dao/internal/store_client_sessions.go | 2 + .../storeDesktopSetting.go | 23 +- internal/model/do/store_client_sessions.go | 1 + .../model/entity/store_client_sessions.go | 1 + internal/packed/packed.go | 2 +- utility/mqtt/emqx/emqx.go | 381 ++++++++++-------- 7 files changed, 252 insertions(+), 173 deletions(-) create mode 100644 internal/consts/emqx.go diff --git a/internal/consts/emqx.go b/internal/consts/emqx.go new file mode 100644 index 0000000..089bc62 --- /dev/null +++ b/internal/consts/emqx.go @@ -0,0 +1,15 @@ +package consts + +// UP 上行消息 cmd 常量 +const ( + CmdMemberLevels = 1 // 会员等级数据 + CmdClientList = 2 // 客户机列表数据 + CmdClientUp = 104 // 上机记录 + CmdClientDown = 106 // 下机记录 +) + +// DOWN 下行消息 cmd 常量 +const ( + CmdDesktopSetting = 10001 // 桌面组件显示设置 + CmdUserFee = 10002 // 用户网费下发 +) diff --git a/internal/dao/internal/store_client_sessions.go b/internal/dao/internal/store_client_sessions.go index 3293c84..d187968 100644 --- a/internal/dao/internal/store_client_sessions.go +++ b/internal/dao/internal/store_client_sessions.go @@ -33,6 +33,7 @@ type StoreClientSessionsColumns struct { CreatedAt string // 创建时间 UpdatedAt string // 更新时间 DeletedAt string // 软删除时间戳 + AreaName string // } // storeClientSessionsColumns holds the columns for the table store_client_sessions. @@ -50,6 +51,7 @@ var storeClientSessionsColumns = StoreClientSessionsColumns{ CreatedAt: "created_at", UpdatedAt: "updated_at", DeletedAt: "deleted_at", + AreaName: "area_name", } // NewStoreClientSessionsDao creates and returns a new DAO object for table data access. diff --git a/internal/logic/storeDesktopSetting/storeDesktopSetting.go b/internal/logic/storeDesktopSetting/storeDesktopSetting.go index 681eada..0cbb8f7 100644 --- a/internal/logic/storeDesktopSetting/storeDesktopSetting.go +++ b/internal/logic/storeDesktopSetting/storeDesktopSetting.go @@ -5,11 +5,13 @@ import ( "encoding/json" "fmt" "github.com/gogf/gf/v2/errors/gerror" + "server/internal/consts" "server/internal/dao" "server/internal/model" "server/internal/model/do" "server/internal/service" "server/utility/mqtt" + "server/utility/mqtt/emqx" ) type sStoreDesktopSetting struct { @@ -63,15 +65,26 @@ func (s *sStoreDesktopSetting) Save(ctx context.Context, in model.SaveDesktopSet if err != nil { return nil, err } - marshal, err := json.Marshal(in) - if err != nil { - return nil, err - } client, b := mqtt.GetClient("emqx") if !b { return nil, gerror.New("获取MQTT客户端失败") } - err = client.Publish(fmt.Sprintf("/desktop/%d", in.StoreId), marshal) + downData := emqx.DownData{ + CMD: consts.CmdDesktopSetting, + StoreId: int(in.StoreId), + Data: struct { + RightComponentVisible int `json:"rightComponentVisible"` + TopComponentVisible int `json:"topComponentVisible"` + }{ + RightComponentVisible: in.RightComponentVisible, + TopComponentVisible: in.TopComponentVisible, + }, + } + marshal, err := json.Marshal(downData) + if err != nil { + return nil, err + } + err = client.Publish(fmt.Sprintf("/%d/down", in.StoreId), marshal) if err != nil { return nil, err } diff --git a/internal/model/do/store_client_sessions.go b/internal/model/do/store_client_sessions.go index fc5dcf1..58e8ffc 100644 --- a/internal/model/do/store_client_sessions.go +++ b/internal/model/do/store_client_sessions.go @@ -25,4 +25,5 @@ type StoreClientSessions struct { CreatedAt *gtime.Time // 创建时间 UpdatedAt *gtime.Time // 更新时间 DeletedAt *gtime.Time // 软删除时间戳 + AreaName interface{} // } diff --git a/internal/model/entity/store_client_sessions.go b/internal/model/entity/store_client_sessions.go index 2842147..d3c11d4 100644 --- a/internal/model/entity/store_client_sessions.go +++ b/internal/model/entity/store_client_sessions.go @@ -23,4 +23,5 @@ type StoreClientSessions struct { CreatedAt *gtime.Time `json:"createdAt" orm:"created_at" description:"创建时间"` // 创建时间 UpdatedAt *gtime.Time `json:"updatedAt" orm:"updated_at" description:"更新时间"` // 更新时间 DeletedAt *gtime.Time `json:"deletedAt" orm:"deleted_at" description:"软删除时间戳"` // 软删除时间戳 + AreaName string `json:"areaName" orm:"area_name" description:""` // } diff --git a/internal/packed/packed.go b/internal/packed/packed.go index e61b667..d483800 100644 --- a/internal/packed/packed.go +++ b/internal/packed/packed.go @@ -4,7 +4,7 @@ import ( _ "github.com/gogf/gf/contrib/drivers/mysql/v2" _ "github.com/gogf/gf/contrib/nosql/redis/v2" _ "server/utility/gamelife" - //_ "server/utility/mqtt/emqx" + _ "server/utility/mqtt/emqx" _ "server/utility/myCasbin" _ "server/utility/oss/aliyun" _ "server/utility/rsa" diff --git a/utility/mqtt/emqx/emqx.go b/utility/mqtt/emqx/emqx.go index ea5354e..93749f7 100644 --- a/utility/mqtt/emqx/emqx.go +++ b/utility/mqtt/emqx/emqx.go @@ -2,11 +2,12 @@ package emqx import ( "context" + "encoding/json" "fmt" - "github.com/gogf/gf/v2/encoding/gjson" - "github.com/gogf/gf/v2/util/gconv" + "github.com/gogf/gf/v2/os/gtime" + "server/internal/consts" "server/internal/dao" - "strconv" + "server/internal/model/do" "sync" "time" @@ -88,12 +89,29 @@ func (e *emqxClient) Subscribe(topic string, handler func(topic string, payload return err } -type DeviceData struct { - NetbarAccount string `json:"netbarAccount"` - DeviceId string `json:"deviceId"` - DeviceName string `json:"deviceName"` - IP string `json:"ip"` - MACAddress string `json:"macAddress"` +type MemberLevelData struct { + LevelID int `json:"level_id"` // 会员等级ID（如0表示临时卡） + LevelName string `json:"level_name"` // 等级名称 + Level int `json:"level"` // 等级顺序，值越大等级越高 + OlType int8 `json:"ol_type"` // 在线类型，例如：1=线上 + Status int8 `json:"status"` // 状态：1=启用，2=禁用 +} + +type ClientData struct { + AreaName string `json:"area_name"` + ClientName string `json:"client_name"` + Status int8 `json:"status"` +} +type ClientSession struct { + AreaName string `json:"area_name"` + ClientName string `json:"client_name"` + CardId string `json:"card_id"` + XyUserId string `json:"xy_user_id"` + Level int `json:"level"` + LevelName string `json:"level_name"` + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` + UsedTime int `json:"used_time"` } // init 注册emqx客户端 @@ -116,181 +134,210 @@ func init() { mqtt.Register("emqx", client) glog.Infof(ctx, "EMQX 客户端注册完成，broker=%s, clientID=%s", broker, clientId) - // 订阅设备上线消息 - // 订阅设备上线消息 + go func() { - ctx := context.Background() - err := client.Subscribe("/+/up", func(topic string, payload []byte) { - glog.Infof(ctx, "收到 MQTT 消息，topic=%s", topic) - - var data DeviceData - if err := gjson.Unmarshal(payload, &data); err != nil { - glog.Errorf(ctx, "[/up] 解析设备信息失败: %v", err) + // 监听数据上报通道 + client.Subscribe("/+/up", func(topic string, payload []byte) { + var base struct { + Cmd int `json:"cmd"` + StoreId int `json:"storeId"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(payload, &base); err != nil { + glog.Errorf(ctx, "解析数据失败，topic=%s，错误：%v", topic, err) return } - - deviceId := data.DeviceId - netbarAccount := data.NetbarAccount - now := time.Now().Unix() - - // Redis 统一 key - onlineDevicesKey := "system_device:online_devices" - lastOnlineKey := fmt.Sprintf("system_device:last_online:%s", deviceId) - deviceInfoKey := fmt.Sprintf("system_device:info:%s", deviceId) - netbarDeviceCountKey := fmt.Sprintf("system_device:netbar_device_count:%s", netbarAccount) - - // 判断设备是否在线 - exists, err := g.Redis().HExists(ctx, onlineDevicesKey, deviceId) - if err != nil { - glog.Errorf(ctx, "查询设备在线状态失败 %s: %v", deviceId, err) - return - } - - // 获取上次上线时间，判断是否断线重连 - lastOnlineStr, err := g.Redis().Get(ctx, lastOnlineKey) - shouldSend := false - needIncr := false - - if exists != 1 { - // 设备不在线，首次上线 - shouldSend = true - needIncr = true - } else if err != nil || lastOnlineStr.IsEmpty() { - // 无上线时间记录，断线重连 - shouldSend = true - needIncr = true - } else { - lastOnline, err := strconv.ParseInt(lastOnlineStr.String(), 10, 64) - if err != nil || now-lastOnline > 10 { - // 超过10秒断线重连 - shouldSend = true - needIncr = true - } else { - // 在线且未断线，不加计数，不发送配置 - shouldSend = false - needIncr = false + switch base.Cmd { + case consts.CmdMemberLevels: + // 当前门店会员等级列表数据 + levels := make([]MemberLevelData, 0) + if err := json.Unmarshal(base.Data, &levels); err != nil { + glog.Errorf(ctx, "解析会员等级数据失败，topic=%s，错误：%v", topic, err) + return } - } - - // 更新上线时间并设置20秒过期 - if _, err := g.Redis().Set(ctx, lastOnlineKey, now); err != nil { - glog.Errorf(ctx, "更新上线时间失败 %s: %v", lastOnlineKey, err) - } - if _, err := g.Redis().Expire(ctx, lastOnlineKey, 20); err != nil { - glog.Errorf(ctx, "设置上线时间过期失败 %s: %v", lastOnlineKey, err) - } - - // 需要时增加在线设备计数 - if needIncr { - if _, err := g.Redis().Incr(ctx, netbarDeviceCountKey); err != nil { - glog.Errorf(ctx, "增加 Netbar 在线设备数失败 %s: %v", netbarDeviceCountKey, err) + saveOrUpdateLevels(ctx, base.StoreId, levels) + glog.Infof(ctx, "成功保存会员等级数据，store_id=%d", base.StoreId) + case consts.CmdClientList: + // 当前门店客户机列表数据 + clients := make([]ClientData, 0) + if err := json.Unmarshal(base.Data, &clients); err != nil { + glog.Errorf(ctx, "解析客户机数据失败，topic=%s，错误：%v", topic, err) + return } - } - - // 更新在线设备集合（时间戳） - if _, err := g.Redis().HSet(ctx, onlineDevicesKey, map[string]interface{}{ - deviceId: now, - }); err != nil { - glog.Errorf(ctx, "更新在线设备集合失败 %s: %v", onlineDevicesKey, err) - return - } - - // 更新设备基础信息，并设置1小时过期 - if _, err := g.Redis().HSet(ctx, deviceInfoKey, map[string]interface{}{ - "netbarAccount": netbarAccount, - "deviceName": data.DeviceName, - "ip": data.IP, - "macAddress": data.MACAddress, - }); err != nil { - glog.Errorf(ctx, "存储设备信息失败 %s: %v", deviceInfoKey, err) - } - if _, err := g.Redis().Expire(ctx, deviceInfoKey, 3600); err != nil { - glog.Errorf(ctx, "设置设备信息过期失败 %s: %v", deviceInfoKey, err) - } - - // 首次或断线重连发送配置 - if shouldSend { - one, err := dao.Stores.Ctx(ctx).InnerJoin(dao.StoreDesktopSettings.Table(), "sds", "sds.store_id = stores.id"). - Where("stores.netbar_account = ?", netbarAccount). - Fields("stores.netbar_account netbarAccount, sds.top_component_visible topComponentVisible, sds.right_component_visible rightComponentVisible").One() - if err != nil { - glog.Errorf(ctx, "获取门店信息失败 %s: %v", deviceInfoKey, err) - } - if err = client.Publish(fmt.Sprintf("/%s/down", netbarAccount), gconv.Bytes(one.Json())); err != nil { - glog.Errorf(ctx, "发布消息失败 %s: %v", deviceInfoKey, err) + saveOrUpdateClients(ctx, base.StoreId, clients) + case consts.CmdClientUp, consts.CmdClientDown: + session := ClientSession{} + if err := json.Unmarshal(base.Data, &session); err != nil { + glog.Errorf(ctx, "解析客户机数据失败，topic=%s，错误：%v", topic, err) + return } + saveOrUpdateClientSession(ctx, base.StoreId, session) } - }) - if err != nil { - glog.Errorf(ctx, "订阅 /+/up 失败: %v", err) - } }() +} - // 监控离线设备 - go func() { - ctx := context.Background() - ticker := time.NewTicker(10 * time.Second) - defer ticker.Stop() +func saveOrUpdateClientSession(ctx context.Context, storeId int, session ClientSession) { + // 查询客户机信息 + client, err := dao.StoreClients.Ctx(ctx). + Where("store_id", storeId). + Where("client_name", session.ClientName). + One() + if err != nil || client.IsEmpty() { + glog.Errorf(ctx, "未找到客户机信息：store_id=%d client_name=%s", storeId, session.ClientName) + return + } + clientId := client["id"].Int64() + areaId := client["area_id"].Int64() - for { - select { - case <-ctx.Done(): - glog.Info(ctx, "停止离线设备监控") - return - case <-ticker.C: - onlineDevicesKey := "system_device:online_devices" - devicesVar, err := g.Redis().HGetAll(ctx, onlineDevicesKey) - if err != nil { - glog.Errorf(ctx, "获取在线设备失败: %v", err) - continue - } + // 上机逻辑（没有 end_time） + if session.EndTime == "" { + // 插入上机记录 + _, err := dao.StoreClientSessions.Ctx(ctx).Data(do.StoreClientSessions{ + StoreId: int64(storeId), + ClientId: clientId, + AreaId: areaId, + XyUserId: session.XyUserId, + LevelId: session.Level, + LevelName: session.LevelName, + StartTime: gtime.NewFromStr(session.StartTime), + AreaName: session.AreaName, + }).Insert() + if err != nil { + glog.Errorf(ctx, "插入上机记录失败：store_id=%d client_id=%d xy_user_id=%s，错误：%v", storeId, clientId, session.XyUserId, err) + return + } + // 更新客户机状态为 上机中（3） + _, err = dao.StoreClients.Ctx(ctx). + Where("id", clientId). + Data(g.Map{"status": 3}). + Update() + if err != nil { + glog.Errorf(ctx, "更新客户机状态为上机中失败：client_id=%d，错误：%v", clientId, err) + } + return + } - devices := devicesVar.MapStrStr() - if len(devices) == 0 { - continue - } + // 下机逻辑（end_time 存在） + _, err = dao.StoreClientSessions.Ctx(ctx). + Where("store_id", storeId). + Where("client_id", clientId). + Where("xy_user_id", session.XyUserId). + Where("start_time", session.StartTime). + Data(g.Map{ + "end_time": session.EndTime, + "used_time": session.UsedTime, + }).Update() + if err != nil { + glog.Errorf(ctx, "更新下机记录失败：store_id=%d client_id=%d xy_user_id=%s，错误：%v", storeId, clientId, session.XyUserId, err) + return + } + // 更新客户机状态为 空闲（1） + _, err = dao.StoreClients.Ctx(ctx). + Where("id", clientId). + Data(g.Map{"status": 1}). + Update() + if err != nil { + glog.Errorf(ctx, "更新客户机状态为空闲失败：client_id=%d，错误：%v", clientId, err) + } +} - now := time.Now().Unix() - for deviceId, timestampStr := range devices { - timestamp, err := strconv.ParseInt(timestampStr, 10, 64) - if err != nil { - glog.Errorf(ctx, "无效时间戳 for 设备 %s: %v", deviceId, err) - continue - } +func saveOrUpdateClients(ctx context.Context, storeId int, clients []ClientData) { + for _, client := range clients { + var areaId int64 + // Step 1: 查询区域 ID + value, err := dao.StoreAreas.Ctx(ctx).Where(do.StoreAreas{StoreId: storeId, AreaName: client.AreaName}).Fields(dao.StoreAreas.Columns().Id).Value() + if err != nil { + glog.Errorf(ctx, "查询区域失败 store_id=%d area_name=%s err=%v", storeId, client.AreaName, err) + continue + } - if now-timestamp > 10 { - // 超过10秒未更新，认定离线 - deviceInfoKey := fmt.Sprintf("system_device:info:%s", deviceId) - dataVar, err := g.Redis().HGetAll(ctx, deviceInfoKey) - if err != nil { - glog.Errorf(ctx, "获取设备数据失败 %s: %v", deviceInfoKey, err) - continue - } + if value.IsEmpty() { + // 不存在则插入数据 + id, err := dao.StoreAreas.Ctx(ctx).Data(do.StoreAreas{ + StoreId: storeId, + AreaName: client.AreaName, + }).InsertAndGetId() + if err != nil { + glog.Errorf(ctx, "插入区域失败 store_id=%d area_name=%s err=%v", storeId, client.AreaName, err) + continue + } + areaId = id + } else { + areaId = value.Int64() + } - data := dataVar.MapStrStr() - netbarAccount, exists := data["netbarAccount"] - if !exists { - glog.Errorf(ctx, "设备 %s 缺少 netbarAccount", deviceId) - continue - } + // Step 2: 查询客户机是否存在 + count, err := dao.StoreClients.Ctx(ctx).Where(do.StoreClients{StoreId: storeId, ClientName: client.ClientName, AreaId: areaId}).Count() + if err != nil { + glog.Errorf(ctx, "查询客户机失败 store_id=%d client_name=%s err=%v", storeId, client.ClientName, err) + continue + } - // 新增：system_device命名空间，按账号统计在线设备数 -1 - netbarDeviceCountKey := fmt.Sprintf("system_device:netbar_device_count:%s", netbarAccount) - if _, err := g.Redis().Decr(ctx, netbarDeviceCountKey); err != nil { - glog.Errorf(ctx, "减少 Netbar 在线设备数失败 %s: %v", netbarDeviceCountKey, err) - } + data := g.Map{ + "store_id": storeId, + "client_name": client.ClientName, + "area_id": areaId, + "status": client.Status, + } - // 从在线设备集合中删除 - if _, err := g.Redis().HDel(ctx, onlineDevicesKey, deviceId); err != nil { - glog.Errorf(ctx, "移除设备 %s 从在线集合失败: %v", deviceId, err) - } else { - glog.Infof(ctx, "设备 %s 已标记为离线", deviceId) - } - } - } + // Step 3: 更新 or 插入 + if count > 0 { + _, err = dao.StoreClients.Ctx(ctx). + Data(data). + Where("store_id", storeId). + Where("client_name", client.ClientName). + Update() + if err != nil { + glog.Errorf(ctx, "更新客户机失败 store_id=%d client_name=%s err=%v", storeId, client.ClientName, err) + } + } else { + _, err = dao.StoreClients.Ctx(ctx). + Data(data). + Insert() + if err != nil { + glog.Errorf(ctx, "插入客户机失败 store_id=%d client_name=%s err=%v", storeId, client.ClientName, err) } } - }() - + } +} + +func saveOrUpdateLevels(ctx context.Context, storeId int, levels []MemberLevelData) { + for _, level := range levels { + exists, err := dao.StoreMemberLevels.Ctx(ctx). + Where("store_id", storeId). + Where("level_id", level.LevelID). + Count() + if err != nil { + glog.Infof(ctx, "检查会员等级是否存在失败，store_id=%d level_id=%d", storeId, level.LevelID) + } + data := g.Map{ + "store_id": storeId, + "level_id": level.LevelID, + "level_name": level.LevelName, + "level": level.Level, + "ol_type": level.OlType, + "status": level.Status, + } + if exists > 0 { + _, err = dao.StoreMemberLevels.Ctx(ctx). + Where("store_id", storeId). + Where("level_id", level.LevelID). + Data(data). + Update() + } else { + _, err = dao.StoreMemberLevels.Ctx(ctx). + Data(data). + Insert() + } + if err != nil { + glog.Errorf(ctx, "保存会员等级数据失败，store_id=%d level_id=%d", storeId, level.LevelID) + } + } +} + +type DownData struct { + CMD int `json:"cmd"` + StoreId int `json:"storeId"` + Data interface{} `json:"data"` }