淄博的大型网站建设,企业网站源码 thinkphp,企业邮箱怎么获取,seo与sem的区别文章目录 前言分析方案1:遗嘱消息演示遗嘱消息的使用实施流程 方案2:使用WebHook开启WebHook演示Webhook编写代码 前言
获取一个设备的在线和离线状态#xff0c;是一个很关键的功能。我们对设备下发的控制指令#xff0c;设备处于在线状态才能及时给我们反馈。这里的在线和… 文章目录 前言分析方案1:遗嘱消息演示遗嘱消息的使用实施流程 方案2:使用WebHook开启WebHook演示Webhook编写代码 前言
获取一个设备的在线和离线状态是一个很关键的功能。我们对设备下发的控制指令设备处于在线状态才能及时给我们反馈。这里的在线和离线我们可以简单的理解为设备与MQTT的连接状态。
分析
我们打电话的时候经常能听到您拨打的用户已关机“和”用户不在服务区或暂时无法接通“这两种的区别是什么 1、当用户开机时会自动向最近的移动基站注册基站标记该用户为attach(在线)状态。 2、当用户关机时手机会发起datach流程告知基站自己关机了基站标记该用户为detach(离线)状态。这样再次拨打就可以节省寻呼资源直接提示用户关机。 3、当用户忽然进入无网络的环境或者手机故障导致来不及发起datach流程基站还认为用户在线,当有人拨打用户号码时基站测会对用户进行寻呼但是超时得不到回应后就会提示不在服务区或者暂时无法接通 的语音。
其实这个方案在IoT上也是可行的我们可以让设备在线和离线的过程中向特定Topic发送状态消息但是存在问题我们需要一个单独的Broker去订阅这个Topic但是这个单独的Broker很容易成为单点故障点。而且如果设备数量很大这种意外离线的设备也很难及时发现需要下发指令后等待设备响应超时才能发现。
方案1:遗嘱消息
MQTT 遗嘱消息可以在客户端意外断线时将“遗嘱”优雅地发送给第三方订阅者以实现离线通知、设备状态更新等业务。其中意外断线指客户端断开前未向服务器发送 DISCONNECT 消息比如
因网络故障或网络波动设备在保持连接周期内未能通讯连接被服务端关闭 设备意外掉电 设备尝试进行不被允许的操作而被服务端关闭连接例如订阅自身权限以外的主题等 遗嘱消息在 MQTT 客户端向服务器端 CONNECT 请求时设置可选属性包括是否发送遗嘱消息 (Will Message)标志和遗嘱消息主题 (Topic) 与内容(Payload) 以及 Properties。
值得一提的遗嘱消息发布的时间可能会有延迟通常意外断线时服务器无法立即检测到断线行为需要通过连接保活心跳机制并经过一定周期后才会触发MQTT 5.0 提供的遗嘱延迟间隔Will Delay Interval属性也会影响发布时间。
演示遗嘱消息的使用
我们使用A、B两台电脑使用MQTT X来演示。 我们在A电脑的 MQTT X 中新建一个名为 Test 的连接Host 修改为 修改为我们的MQTT地址192.120.5.204并输入账号密码在 Advanced 部分选择 MQTT Version 为 5.0并且将 Session Expiry Interval 设置为 10确保会话不会在遗嘱消息发布前过期。 然后在 Lass Will and Testament 部分将 Last-Will Topic 设置为 offlineLast-Will Payload 设置为 I’m offlineWill Delay Interval (s) 设置为 5。 完成以上设置后我们点击右上角的 Connect 按钮以建立连接。
我们在B电脑的MQTTX中新建一个连接Submqtt地址同样指向我们的mqtt服务器192.120.5.204
并订阅offline主题 我们用任务管理器直接结束A电脑的MQTTX进程这是连接会被直接断开模拟了设备断电的场景在5s之后在B电脑的MQTTX订阅中收到了一条内容为 I‘m offline 的遗嘱消息。
实施流程
1、设备遗嘱消息内容设置为offline该遗嘱主题与一个普通发送状态的主题设定成同一个 {设备名称}/status。例如 284202304230001/status 2、当设备连接时向主题 {设备名称}/status 发送内容为 online 的Retained消息其它客户端订阅主题 {设备名称}/status 的时候将获取到 Retained 消息为 online。 保留消息(Retain ) MQTT 服务端收到 Retain 标志为 1 的 PUBLISH 报文时会将该报文视为保留消息除了被正常转发以外 保留消息会被存储在服务端每个主题下只能存在一份保留消息因此如果已经存在相同主题的保留消息则该保留消息被替换。 当客户端建立订阅时如果服务端存在主题匹配的保留消息则这些保留消息将被立即发送给该客户端。 借助保留消息新的订阅者能够立即获取最近的状态而不需要等待无法预期的时间这在很多场景下是非常重要的。 EMQX 默认开启保留消息的能力和服务可以在 etc/emqx.conf 中修改 mqtt.retain_available 为 false 来关闭保留消息的能力 这样客户端将被禁止发送 Retain 标志为 1 的 PUBLISH 报文否则客户端将会收到原因码为 0x9A不支持保留消息的 DISCONNECT 报文。 保留消息的服务会存储和管理客户端发送的保留消息并发送给相应的订阅者。 3、当客户端异常断开时系统自动向主题 {设备名称}/status 发送内容为 offline 的消息其它订阅了此主题的客户端会马上收到 offline 消息如果遗嘱消息设置了 Will Retain那么此时如果有新的订阅 A/status 主题的客户端上线也将获取到内容为 offline 的遗嘱消息。
方案2:使用WebHook
方案1需要设备主动设置遗嘱消息才能实现那么有没有更简单的方式直接通过设备与Mqtt的连接事件来获取连接状态呢。 EMQX 设计了一套WebHook系统可以通过这个自带的WebHook系统获取内部的事件并进行处理。WebHook的原理很简单当设备与mqtt建立连接或者断开连接时EMQX会把事件的信息通过我们的配置调用特定的URL上的接口实现通知。 使用WebHook还可以有限避免单点故障。所以本项目会采用WebHook的方式来实现对设备在线和离线的管理。
开启WebHook
在数据集成 - 数据桥接 中创建一个Webhook 名称设置为ConnectedEvent,URL 中填写我们的Webhook地址也就是触发事件之后的调用接口地址这里我们填 http://192.120.5.204:5000/api/Device/ConnectedEvent 请求方式为Post其他内容保持默认不变 这里注意URL可以通过${field}的方式拼接请求体也可以自己指定如果留空会原样转发消息我们这里请求体留空 设备在线和离线的事件转发的消息格式如下
{username: 284202304230001,timestamp: 1682652598840,sockname: 172.17.0.5:1883,receive_maximum: 32,proto_ver: 5,proto_name: MQTT,peername: 172.17.0.1:48524,node: emqx172.17.0.5,mountpoint: undefined,metadata: {rule_id: rule_3hsx},keepalive: 60,is_bridge: false,expiry_interval: 10,event: client.connected,connected_at: 1682652598840,conn_props: {User-Property: {},Session-Expiry-Interval: 10},clientid: mqttx_c4491df0,clean_start: false
}
我们点击 创建 并继续点击 创建规则 我们在创建规则中指定新的规则名称 rule_client_connected,并在SQL编辑器复制以下内容
SELECT*
FROM$events/client_connected,$events/client_disconnected在右侧的事件中我们可以看到所有可用的事件我们选择了连接和断开两个事件在这两个事件触发时会通过Webhook调用我们配置的接口这样我们就能获取到设备的在线、离线状态了。 我们点击 创建按钮 完成规则的创建 我们可以看见我们创建好的规则 点击规则ID还可以看到统计数据
在FLows中还可以看到整个工作流程
演示Webhook
我们使用MQTTX模拟一次设备连接和断开动作可以在规则统计界面看到我们的操作已经被记录。
编写代码
我们这里采用方案2。 我们需要实现之前配置的ConnectedEvent接口 /// summary/// 连接事件请求/// /summarypublic class ConnectedEventRequest{/// summary/// 设备名称/// /summarypublic string Username { get; set; }/// summary/// 时间戳/// /summarypublic long Timestamp { get; set; }/// summary/// 事件连接/断开/// /summarypublic string Event { get; set; }/// summary/// 连接时间断开事件中为0/// /summarypublic long Connected_at { get; set; }/// summary/// Client ID/// /summarypublic string Clientid { get; set; }}/// summary/// 更新设备在线状态/// /summary/// param namedeviceName/param/// param nameonlineStatus/param/// returns/returnspublic async Task UpdateDeviceOnlineStatusAsync(string deviceName, OnLineStates onlineStatus){var device await _ioTDbContext.IoTDeviceInfo.Include(o o.IoTDeviceExtend).AsNoTracking().FirstOrDefaultAsync(o o.DeviceName deviceName);if (device null){return;}else{if (device.IoTDeviceExtend null) //扩展表为空{device.IoTDeviceExtend new IoTDeviceExtend{DeviceInfoId device.Id,OnLineStates (int)onlineStatus,};_ioTDbContext.Attach(device.IoTDeviceExtend);_ioTDbContext.Entry(device.IoTDeviceExtend).State EntityState.Added;_ioTDbContext.Entry(device.IoTDeviceExtend).Property(o o.OnLineStates).IsModified true;await _ioTDbContext.SaveChangesAsync();}if (device.IoTDeviceExtend.OnLineStates ! (int)onlineStatus) //在线状态不一致{device.IoTDeviceExtend.OnLineStates (int)onlineStatus;_ioTDbContext.Attach(device.IoTDeviceExtend);//防止更新其他字段_ioTDbContext.Entry(device.IoTDeviceExtend).State EntityState.Unchanged;_ioTDbContext.Entry(device.IoTDeviceExtend).Property(o o.OnLineStates).IsModified true;await _ioTDbContext.SaveChangesAsync();}}}我们根据Event中的内容来判断是 连接client.connected/断开(client.disconnected) 的事件 /// summary/// 连接、断开事件/// /summary/// param namerequest/param/// returns/returns[HttpPost]public async Task ConnectedEventAsync([FromBody] ConnectedEventRequest request){var onlineStatus request.Event switch{client.connected OnLineStates.OnLine,_ OnLineStates.OffLine};await _deviceHandler.UpdateDeviceOnlineStatusAsync(request.Username, onlineStatus);}#总结 以上就是本文要讲的内容我们可以通过MQTTX来测试我们的代码有效性。 该方案还存在部分缺点例如 1、每次设备上下线会导致频繁的请求接口在大量设备接入的场景中需要考虑接口性能。 2、由于网络等问题Web调用顺序可能不能完全保证也许离线会比在线事件更早处理从而导致状态不一致。我们后面会尝试用其他方案来替代WebHook尝试解决上述问题在此之前我们都会继续使用WebHook进行功能演示。
完整代码在这里https://github.com/sunday866/MASA.IoT-Training-Demos
如果你对我们的 MASA 感兴趣无论是代码贡献、使用、提 Issue欢迎联系我们
WeChatMasaStackTechOps QQ7424099
