更新

JiShe.ServicePro

@@ -1 +1 @@
-Subproject commit 3dbd0ebf5fff82ae03945631aa8e401620931ab1
+Subproject commit 2cc68d62afbcde8e8cc7bfdc57c25ecf6fcbc54e

host/JiShe.IoT.HttpApi.Host/appsettings.Development.json

@@ -32,10 +32,11 @@
     "TaskThreadCount": -1
   },
   "Pulsar": {
-    "ServiceUrl": "pulsar://192.168.56.101:6650",
-    "WebUrl": "http://192.168.56.101:8080",
+    "ServiceUrl": "pulsar://47.110.60.222:9093",
+    "WebUrl": "http://47.110.60.222:9094",
+    "UserName": "admin",
     "TenantName": "1YMVZZkAkRArjxSD8457",
-    "Namespace": "default",
+    "Namespace": "OneNET",
     "PulsarSecretKey": "0fd7afb8b0d04e6abc4fdfdac2190a79",
     "PulsarSubscriptionCustomName": "sub",
     "EnableTls": false,
@@ -58,8 +59,10 @@
   "IoTDBOptions": {
     "UserName": "root",
     "Password": "Lixiao@1980",
-    "TreeModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
-    "TableModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
+    //"TreeModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
+    //"TableModelClusterList": [ "47.110.53.196:6667", "47.110.60.222:6667", "47.110.62.104:6667" ],
+    "TreeModelClusterList": [ "47.110.53.196:30710" ],
+    "TableModelClusterList": [ "47.110.53.196:30710" ],
     "PoolSize": 32,
     "DataBaseName": "jisheiotdata",
     "OpenDebugMode": true,

src/JiShe.IoT.Application.Contracts/Subscribers/IServiceCommunicationChannelSubscriberService.cs

@@ -1,7 +1,4 @@
-using JiShe.ServicePro.Dto;
-using JiShe.ServicePro.Kafka.Internal; 
-
-namespace JiShe.ServicePro.OneNETManagement.Subscribers
+namespace JiShe.ServicePro.OneNETManagement.Subscribers
 {
     /// <summary>
     /// 数据通讯通道订阅者服务
@@ -9,10 +6,9 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
     public interface IServiceCommunicationChannelSubscriberService
     {
         /// <summary>
-        /// 设备上下线通知Kafka消息订阅
+        /// 设备上下线通知 Redis 消息订阅
         /// </summary> 
-        /// <param name="message"></param>
         /// <returns></returns>
-        Task<ISubscribeAck> ServiceCommunicationDeviceStatusSubscriber(DeviceStatusMessage message);
+        Task ServiceCommunicationDeviceStatusSubscriber();
     }
 }

src/JiShe.IoT.Application/IoTApplicationModule.cs

@@ -8,6 +8,7 @@ using JiShe.ServicePro.FreeRedisProvider;
 using JiShe.ServicePro.FreeSqlProvider;
 using JiShe.ServicePro.IoTDBManagement;
 using JiShe.ServicePro.OneNETManagement;
+using JiShe.ServicePro.OneNETManagement.Subscribers;
 using JiShe.ServicePro.TemplateManagement;
 using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp;
@@ -58,6 +59,9 @@ namespace JiShe.IoT
         {
             var commonService = context.ServiceProvider.GetRequiredService<ICommonService>();
             commonService.InitSelectTypetList();
+
+            var communicationChannelService = context.ServiceProvider.GetRequiredService<IServiceCommunicationChannelSubscriberService>();
+            communicationChannelService.ServiceCommunicationDeviceStatusSubscriber();
         }
     }
 }

src/JiShe.IoT.Application/Subscribers/ServiceCommunicationChannelSubscriberService.cs

@@ -1,4 +1,6 @@
-using JiShe.IoT;
+using FreeRedis;
+using FreeSql;
+using JiShe.IoT;
 using JiShe.ServicePro.ApacheIoTDB.Provider.Model;
 using JiShe.ServicePro.ApacheIoTDB.Provider.Options;
 using JiShe.ServicePro.Consts;
@@ -8,6 +10,7 @@ using JiShe.ServicePro.DeviceManagement.DeviceInfos;
 using JiShe.ServicePro.Dto;
 using JiShe.ServicePro.Enums;
 using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.FreeSqlProvider;
 using JiShe.ServicePro.IoTDBManagement.DataChannels;
 using JiShe.ServicePro.IoTDBManagement.TableModels;
 using JiShe.ServicePro.Kafka.Attributes;
@@ -25,51 +28,85 @@ using Volo.Abp.Auditing;
 namespace JiShe.ServicePro.OneNETManagement.Subscribers
 {
     /// <summary>
-    /// 设备状态统一订阅订阅服务
+    /// 数据通讯通道订阅者服务
     /// </summary>
     [DisableAuditing]
-    public class ServiceCommunicationChannelSubscriberService : IoTAppService, IServiceCommunicationChannelSubscriberService, IKafkaSubscribe
+    public class ServiceCommunicationChannelSubscriberService : IoTAppService, IServiceCommunicationChannelSubscriberService
     {
         private readonly ILogger<ServiceCommunicationChannelSubscriberService> _logger;
         private readonly IoTDBOptions _iotDBOptions;
+        private readonly IRedisPubSubService _redisPubSubService;
 
         /// <summary>
         /// 设备状态统一订阅订阅服务
         /// </summary>
         /// <param name="logger"></param>
         /// <param name="iotDBOptions"></param>
-        public ServiceCommunicationChannelSubscriberService(ILogger<ServiceCommunicationChannelSubscriberService> logger, IOptions<IoTDBOptions> iotDBOptions)
+        /// <param name="redisPubSubService"></param>
+        public ServiceCommunicationChannelSubscriberService(ILogger<ServiceCommunicationChannelSubscriberService> logger, IOptions<IoTDBOptions> iotDBOptions, IRedisPubSubService redisPubSubService)
         {
             _logger = logger;
             _iotDBOptions = iotDBOptions.Value;
+            _redisPubSubService = redisPubSubService;
         }
 
         /// <summary>
-        /// 设备上下线通知Kafka消息订阅
+        /// 设备上下线通知 Redis 消息订阅
         /// </summary>
-        /// <param name="message"></param>
         /// <returns></returns>
-        [KafkaSubscribe(KafkaTopicConsts.ServiceCommunicationDeviceStatusEventName)]
-        public async Task<ISubscribeAck> ServiceCommunicationDeviceStatusSubscriber(DeviceStatusMessage message)
+        public async Task ServiceCommunicationDeviceStatusSubscriber()
         {
             try
             {
-                var deviceEntity = await FreeSqlDbContext.Instance.Select<DeviceManagementInfo>()
-                    .Where(d => d.IoTPlatform == message.IoTPlatform && d.DeviceAddress == message.DeviceAddress)
+                // 为订阅回调创建独立的 FreeSql 客户端
+                var callbackFreeSqlDbContext = FreeSqlDbContext;
+                // 订阅频道
+                await _redisPubSubService.SubscribeAsync<DeviceStatusMessage>(RedisConst.ServiceCommunicationDeviceStatusEventName,  (message) =>
+                {
+                    _logger.LogWarning($"Redis订阅收到设备状态消息: {message}");
+                      
+                    HandDeviceStatus(message, callbackFreeSqlDbContext).ConfigureAwait(false).GetAwaiter().GetResult();
+                });
+            }
+            catch (Exception)
+            {
+
+                throw;
+            }
+        }
+
+        /// <summary>
+        /// 设备状态处理
+        /// </summary>
+        /// <param name="deviceStatusMessage"></param>
+        /// <param name="callbackFreeSqlDbContext"></param>
+        /// <returns></returns>
+        /// <exception cref="UserFriendlyException"></exception>
+        private async Task HandDeviceStatus(DeviceStatusMessage deviceStatusMessage, IFreeSqlProvider callbackFreeSqlDbContext)
+        {
+            try
+            {
+                if (deviceStatusMessage == null || callbackFreeSqlDbContext == null)
+                {
+                    throw new UserFriendlyException($"设备状态消息处理失败，参数为空");
+                }
+
+                var deviceEntity = await callbackFreeSqlDbContext.Instance.Select<DeviceManagementInfo>()
+                    .Where(d => d.IoTPlatform == deviceStatusMessage.IoTPlatform && d.DeviceAddress == deviceStatusMessage.DeviceAddress)
                     .FirstAsync();
                 if (deviceEntity == null)
                 {
-                    return SubscribeAck.Fail();
+                    throw new UserFriendlyException($"{deviceStatusMessage.DeviceAddress} 设备不存在");
                 }
 
-                deviceEntity.DeviceOnlineStatus = message.Status;
-                if (message.Status == DeviceOnlineStatusEnum.Online)
+                deviceEntity.DeviceOnlineStatus = deviceStatusMessage.Status;
+                if (deviceStatusMessage.Status == DeviceOnlineStatusEnum.Online)
                 {
-                    deviceEntity.LastOnlineTime = TimestampHelper.ConvertToDateTime(message.ReceivedTime, TimestampUnit.Milliseconds,DateTimeKind.Local);
+                    deviceEntity.LastOnlineTime = TimestampHelper.ConvertToDateTime(deviceStatusMessage.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
                 }
-                else if (message.Status == DeviceOnlineStatusEnum.Offline)
+                else if (deviceStatusMessage.Status == DeviceOnlineStatusEnum.Offline)
                 {
-                    deviceEntity.LastOfflineTime = TimestampHelper.ConvertToDateTime(message.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
+                    deviceEntity.LastOfflineTime = TimestampHelper.ConvertToDateTime(deviceStatusMessage.ReceivedTime, TimestampUnit.Milliseconds, DateTimeKind.Local);
                 }
 
                 var updateResult = await FreeSqlDbContext.Instance.Update<DeviceManagementInfo>()
@@ -77,10 +114,11 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
                     .UpdateColumns(a => new { a.DeviceOnlineStatus, a.LastOnlineTime, a.LastOfflineTime })
                     .ExecuteAffrowsAsync();
 
-                if (updateResult <=0)
+                if (updateResult <= 0)
                 {
-                    _logger.LogError($"{nameof(ServiceCommunicationDeviceStatusSubscriber)} 设备状态更新失败{message.Serialize()}");
-                    return SubscribeAck.Fail();
+                    _logger.LogError($"{nameof(ServiceCommunicationDeviceStatusSubscriber)} 设备状态更新失败{deviceStatusMessage.Serialize()}");
+
+                    throw new UserFriendlyException($"设备状态更新失败");
                 }
 
                 //更新设备数据缓存到Redis
@@ -89,8 +127,6 @@ namespace JiShe.ServicePro.OneNETManagement.Subscribers
                 deviceCacheInfos.PlatformPassword = null;
 
                 RedisProvider.Instance.HSet<DeviceCacheInfos>(RedisConst.CacheAllDeviceInfoHashKey, deviceEntity.DeviceAddress, deviceCacheInfos);
-
-                return SubscribeAck.Success();
             }
             catch (Exception)
             {