修复了一些BUG

“wangming”
1 parent de2bd2f9
Showing 10 changed files with 168 additions and 18 deletions
.DS_Store
netcore/.DS_Store
netcore/src/.DS_Store
netcore/src/Application/.DS_Store
netcore/src/Application/NCC.API/.DS_Store
netcore/src/Modularity/Extend/.DS_Store
netcore/src/Modularity/Extend/NCC.Extend.Entitys/uavDevice/Dto/UavDeviceCrInput.cs
netcore/src/Modularity/Extend/NCC.Extend/.DS_Store
netcore/src/Modularity/Extend/NCC.Extend/MqttPublisherService.cs
netcore/src/Modularity/Extend/NCC.Extend/NCC.Extend.xml
@@ -48,6 +48,11 @@ namespace NCC.Extend.Entitys.Dto.UavDevice
         /// 备注
         /// </summary>
         public string remark { get; set; }
+
+        /// <summary>
+        /// 区县、位置
+        /// </summary>
+        public string district { get; set; }
         /// <summary>
         /// 机柜货道列表
@@ -69,24 +69,27 @@ public class MqttPublisherService : IMqttPublisherService, ITransient
         });
         // 连接断开事件
-        _mqttClient.UseDisconnectedHandler(e =>
+        _mqttClient.UseDisconnectedHandler(async e =>
         {
             Log.Warning("MQTT 已断开连接");
             int retryInterval = 5; // 秒
-            // while (!_mqttClient.IsConnected)
-            // {
-            //     try
-            //     {
-            //         Log.Information($"尝试在 {retryInterval} 秒后重新连接 MQTT...");
-            //         Task.Delay(TimeSpan.FromSeconds(retryInterval));
-            //         _mqttClient.ConnectAsync(_mqttOptions, CancellationToken.None);
-            //         Log.Information("MQTT 重连成功");
-            //     }
-            //     catch (Exception ex)
-            //     {
-            //         Log.Error($"MQTT 重连失败: {ex.Message}");
-            //     }
-            // }
+            while (!_mqttClient.IsConnected)
+            {
+                try
+                {
+                    Log.Information($"尝试在 {retryInterval} 秒后重新连接 MQTT...");
+                    await Task.Delay(TimeSpan.FromSeconds(retryInterval));
+                    await _mqttClient.ConnectAsync(_mqttOptions, CancellationToken.None);
+                    Log.Information("MQTT 重连成功");
+                    break; // 重连成功后跳出循环
+                }
+                catch (Exception ex)
+                {
+                    Log.Error($"MQTT 重连失败: {ex.Message}");
+                    // 重连失败后继续循环，但增加延迟避免频繁重试
+                    await Task.Delay(TimeSpan.FromSeconds(retryInterval * 2));
+                }
+            }
         });
         // 收到消息事件：处理设备回传的消息
         _mqttClient.UseApplicationMessageReceivedHandler(e =>
@@ -123,17 +126,23 @@ public class MqttPublisherService : IMqttPublisherService, ITransient
                         var lastLog = _db.Queryable<UavMqttMessageLogEntity>().Where(p => p.DeviceId == deviceId && p.Payload == payload && p.CreatedAt >= twoSecondsAgo).Any();
                         if (lastLog == false)
                         {
-
+                            Log.Information($"[订阅消息] 处理新消息：设备 {deviceId}，内容：{payload}");
                             var orderService = _serviceProvider.GetRequiredService<IUavOrderService>();
                             orderService.HandleSubscribeMessage(payload);
                         }
                         else
                         {
-                            Log.Information($"[订阅消息] 接收到重复的消息，已忽略：{payload}");
+                            Log.Information($"[订阅消息] 接收到重复的消息，已忽略：设备 {deviceId}，内容：{payload}");
                         }
                         //只保存正常消息
                         _db.Insertable(logEntity).ExecuteCommand();
                     }
+                    else
+                    {
+                        Log.Debug($"[订阅消息] 接收到确认消息，不处理：设备 {deviceId}，内容：{payload}");
+                        // 保存确认消息到日志
+                        _db.Insertable(logEntity).ExecuteCommand();
+                    }
                 }
             }
             catch (Exception ex)
@@ -157,7 +166,22 @@ public class MqttPublisherService : IMqttPublisherService, ITransient
             // 如果未连接则先连接
             if (!_mqttClient.IsConnected)
             {
+                Log.Warning($"MQTT 未连接，尝试重新连接...");
                 await _mqttClient.ConnectAsync(_mqttOptions, CancellationToken.None);
+                
+                // 等待连接稳定
+                int retryCount = 0;
+                while (!_mqttClient.IsConnected && retryCount < 3)
+                {
+                    await Task.Delay(1000);
+                    retryCount++;
+                }
+                
+                if (!_mqttClient.IsConnected)
+                {
+                    Log.Error("MQTT 连接失败，无法发送消息");
+                    return false;
+                }
             }
             // 构建 MQTT 消息
@@ -189,6 +213,29 @@ public class MqttPublisherService : IMqttPublisherService, ITransient
         catch (Exception ex)
         {
             Log.Error($"MQTT 发送失败：{ex.Message}");
+            
+            // 记录失败的消息到数据库
+            try
+            {
+                var topicParts = topic.Split('/');
+                UavMqttMessageLogEntity logEntity = new UavMqttMessageLogEntity
+                {
+                    Id = YitIdHelper.NextId().ToString(),
+                    MessageType = "send",
+                    Topic = topic,
+                    Payload = payload,
+                    DeviceId = topicParts.Length > 1 ? topicParts[1] : "unknown",
+                    Processed = 0,
+                    Status = "failed",
+                    CreatedAt = DateTime.Now,
+                };
+                _db.Insertable(logEntity).ExecuteCommand();
+            }
+            catch (Exception logEx)
+            {
+                Log.Error($"记录失败消息到数据库时出错：{logEx.Message}");
+            }
+            
             return false;
         }
     }
@@ -454,7 +501,93 @@ public class MqttPublisherService : IMqttPublisherService, ITransient
     #region 私有回调逻辑
+    #endregion
+    /// <summary>
+    /// 检查MQTT连接健康状态
+    /// </summary>
+    /// <returns>连接状态信息</returns>
+    public async Task<dynamic> GetConnectionHealthAsync()
+    {
+        try
+        {
+            if (_mqttClient.IsConnected)
+            {
+                try
+                {
+                    await _mqttClient.PingAsync(CancellationToken.None);
+                    Log.Debug("MQTT Ping 成功");
+                    return new
+                    {
+                        IsConnected = true,
+                        ConnectionTime = DateTime.Now,
+                        LastPing = DateTime.Now,
+                        ClientId = _mqttOptions.ClientId,
+                        BrokerAddress = "mqtt.cqjiangzhichao.cn:1883"
+                    };
+                }
+                catch (Exception pingEx)
+                {
+                    Log.Warning($"MQTT Ping 失败: {pingEx.Message}");
+                    return new
+                    {
+                        IsConnected = false,
+                        ConnectionTime = (DateTime?)null,
+                        LastPing = DateTime.Now,
+                        ClientId = _mqttOptions.ClientId,
+                        BrokerAddress = "mqtt.cqjiangzhichao.cn:1883",
+                        PingError = pingEx.Message
+                    };
+                }
+            }
+            else
+            {
+                return new
+                {
+                    IsConnected = false,
+                    ConnectionTime = (DateTime?)null,
+                    LastPing = DateTime.Now,
+                    ClientId = _mqttOptions.ClientId,
+                    BrokerAddress = "mqtt.cqjiangzhichao.cn:1883"
+                };
+            }
+        }
+        catch (Exception ex)
+        {
+            Log.Error($"获取连接健康状态失败: {ex.Message}");
+            return new
+            {
+                IsConnected = false,
+                Error = ex.Message,
+                Timestamp = DateTime.Now
+            };
+        }
+    }
-    #endregion
+    /// <summary>
+    /// 强制重新连接MQTT
+    /// </summary>
+    /// <returns>重连是否成功</returns>
+    public async Task<bool> ForceReconnectAsync()
+    {
+        try
+        {
+            if (_mqttClient.IsConnected)
+            {
+                await _mqttClient.DisconnectAsync();
+                Log.Information("MQTT 主动断开连接");
+            }
+
+            await Task.Delay(2000); // 等待2秒后重连
+            
+            await _mqttClient.ConnectAsync(_mqttOptions, CancellationToken.None);
+            Log.Information("MQTT 强制重连成功");
+            return true;
+        }
+        catch (Exception ex)
+        {
+            Log.Error($"MQTT 强制重连失败: {ex.Message}");
+            return false;
+        }
+    }
 }
 \ No newline at end of file
@@ -169,6 +169,18 @@
             清理过期的缓存数据（建议定期调用）
             </summary>
         </member>
+        <member name="M:NCC.Extend.MqttPublisherService.GetConnectionHealthAsync">
+            <summary>
+            检查MQTT连接健康状态
+            </summary>
+            <returns>连接状态信息</returns>
+        </member>
+        <member name="M:NCC.Extend.MqttPublisherService.ForceReconnectAsync">
+            <summary>
+            强制重新连接MQTT
+            </summary>
+            <returns>重连是否成功</returns>
+        </member>
         <member name="T:NCC.Extend.UavAdvertisement.UavAdvertisementService">
             <summary>
             广告管理服务