U8调整

ZR.Common/MqttHelper/MqttService.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -164,10 +165,19 @@ namespace ZR.Common.MqttHelper
             }
         }
 
-        private Task OnConnectedAsync(MqttClientConnectedEventArgs e)
+        private async Task OnConnectedAsync(MqttClientConnectedEventArgs e)
         {
-            _logger.LogInformation("MQTT连接已建立");
-            return SubscribeToTopicsAsync();
+            _logger.LogInformation($"MQTT连接已建立，会话是否存在: {e.ConnectResult.IsSessionPresent}");
+
+            // 仅在会话不存在时订阅（首次连接或会话失效）
+            if (!e.ConnectResult.IsSessionPresent)
+            {
+                await SubscribeToTopicsAsync();
+            }
+            else
+            {
+                _logger.LogInformation("会话已存在，服务器保留订阅状态，跳过订阅");
+            }
         }
 
         private async Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs e)
@@ -228,43 +238,47 @@ namespace ZR.Common.MqttHelper
             }
         }
 
+        private readonly HashSet<string> _subscribedTopics = new HashSet<string>(); // 线程安全可加锁
+
         private async Task SubscribeToTopicsAsync()
         {
             if (!_mqttClient.IsConnected)
             {
-                _logger.LogWarning("无法订阅主题：MQTT客户端未连接");
+                _logger.LogWarning("无法订阅：客户端未连接");
                 return;
             }
 
-            try
+            var subscribeOptions = _mqttConfig.GetSubscribeToTopicsAsync();
+            if (subscribeOptions == null || !subscribeOptions.TopicFilters.Any())
             {
-                // 获取配置类生成的订阅选项
-                var subscribeOptions = _mqttConfig.GetSubscribeToTopicsAsync();
-
-                if (subscribeOptions == null || !subscribeOptions.TopicFilters.Any())
-                {
-                    _logger.LogInformation("没有配置要订阅的MQTT主题");
-                    return;
-                }
-
-                _logger.LogInformation(
-                    $"正在订阅MQTT主题: {string.Join(", ", subscribeOptions.TopicFilters.Select(f => f.Topic))}"
-                );
-
-                // 使用强类型的订阅选项进行订阅
-                var result = await _mqttClient.SubscribeAsync(subscribeOptions);
-
-                foreach (var item in result.Items)
-                {
-                    _logger.LogInformation(
-                        $"订阅主题 '{item.TopicFilter.Topic}' 结果: {item.ResultCode}"
-                    );
-                }
+                _logger.LogInformation("无订阅主题配置");
+                return;
             }
-            catch (Exception ex)
+
+            // 提取需要订阅的主题
+            var topicsToSubscribe = subscribeOptions.TopicFilters.Select(f => f.Topic).ToList();
+
+            // 检查是否已订阅所有主题，避免重复
+            if (_subscribedTopics.SetEquals(topicsToSubscribe))
             {
-                _logger.LogError(ex, "订阅MQTT主题时出错");
-                ScheduleReconnect();
+                _logger.LogInformation("所有主题已订阅，跳过订阅操作");
+                return;
+            }
+
+            // 执行订阅并更新本地状态
+            _logger.LogInformation($"订阅主题: {string.Join(", ", topicsToSubscribe)}");
+            var result = await _mqttClient.SubscribeAsync(subscribeOptions);
+
+            // 更新本地已订阅主题列表
+            lock (_subscribedTopics)
+            {
+                _subscribedTopics.Clear();
+                _subscribedTopics.UnionWith(topicsToSubscribe);
+            }
+
+            foreach (var item in result.Items)
+            {
+                _logger.LogInformation($"订阅结果：{item.TopicFilter.Topic} -> {item.ResultCode}");
             }
         }
 
@@ -396,7 +410,6 @@ namespace ZR.Common.MqttHelper
                 _logger.LogError(ex, "清理MQTT客户端资源时出错");
             }
         }
-
         public void Dispose()
         {
             Dispose(true);

ZR.Common/MqttHelper/MyMqttConfig.cs

@@ -40,7 +40,7 @@ namespace ZR.Common.MqttHelper
             var builder = new MqttClientOptionsBuilder()
                 .WithClientId(mqttConfig["ClientId"] ?? Guid.NewGuid().ToString())
                 .WithTcpServer(mqttConfig["Server"], mqttConfig.GetValue<int>("Port", 1883))
-                .WithCleanSession();
+                .WithCleanSession(false);
             return builder.Build();
         }