Merge branch 'feature_定时抄读_05_CY' into dev

src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs

@@ -103,6 +103,7 @@ public class CollectBusApplicationModule : AbpModule
 
         //默认初始化表计信息
         dbContext.InitAmmeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult(); 
+        dbContext.InitWatermeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult();
     }
 
 }

src/JiShe.CollectBus.Application/Plugins/TcpMonitor.cs

@@ -225,9 +225,9 @@ namespace JiShe.CollectBus.Plugins
             //});
 
             
-            string topicName = string.Format(ProtocolConst.AFNTopicNameFormat, aFn);
+            //string topicName = string.Format(ProtocolConst.AFNTopicNameFormat, aFn);
 
-            await _producerBus.PublishAsync(topicName, new MessageReceived
+            await _producerBus.PublishAsync(ProtocolConst.SubscriberReceivedEventName, new MessageReceived
             {
                 ClientId = client.Id,
                 ClientIp = client.IP,

src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs

@@ -184,6 +184,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                         continue;
                     }
 
+                    focusAddressDataList.Add(item.Key);
+
                     var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}{item.Key}";
 
 #if DEBUG
@@ -255,6 +257,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 await FreeRedisProvider.Instance.SetAsync(taskRedisCacheKey, nextTask);
             }
 
+            //初始化设备组负载控制
+            if (focusAddressDataList == null || focusAddressDataList.Count <= 0)
+            {
+                _logger.LogError($"{nameof(InitAmmeterCacheData)} 初始化设备组负载控制失败，没有找到对应的设备信息");
+
+            }
+            else
+            {
+                DeviceGroupBalanceControl.InitializeCache(focusAddressDataList);
+            }
+
             _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成");
         }
 

src/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs

@@ -4,6 +4,7 @@ using System.Threading.Tasks;
 using DotNetCore.CAP;
 using JiShe.CollectBus.Ammeters;
 using JiShe.CollectBus.Common.Consts;
+using JiShe.CollectBus.Common.Helpers;
 using JiShe.CollectBus.FreeSql;
 using JiShe.CollectBus.GatherItem;
 using JiShe.CollectBus.IoTDBProvider;
@@ -15,6 +16,7 @@ using JiShe.CollectBus.Repository;
 using JiShe.CollectBus.Repository.MeterReadingRecord;
 using MassTransit;
 using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
 using Volo.Abp.Domain.Repositories;
@@ -169,5 +171,27 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             .Ado
             .QueryAsync<WatermeterInfo>(sql);
         }
+
+
+        /// <summary>
+        /// 测试设备分组均衡控制算法
+        /// </summary>
+        /// <param name="deviceCount"></param>
+        /// <returns></returns>
+        [HttpGet]
+        public async Task TestDeviceGroupBalanceControl(int deviceCount = 200000)
+        {
+            var deviceList = new List<string>();
+            for (int i = 0; i < deviceCount; i++)
+            {
+                deviceList.Add($"Device_{Guid.NewGuid()}");
+            }
+
+
+            // 打印分布统计
+            DeviceGroupBalanceControl.PrintDistributionStats();
+
+            await Task.CompletedTask;
+        }
     }
 }

src/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs

@@ -7,6 +7,7 @@ using JiShe.CollectBus.Common.Enums;
 using JiShe.CollectBus.Common.Extensions;
 using JiShe.CollectBus.Common.Helpers;
 using JiShe.CollectBus.Common.Models;
+using JiShe.CollectBus.IoTDBProvider;
 using JiShe.CollectBus.IotSystems.Devices;
 using JiShe.CollectBus.IotSystems.MessageReceiveds;
 using JiShe.CollectBus.IotSystems.MeterReadingRecords;
@@ -31,6 +32,7 @@ namespace JiShe.CollectBus.Subscribers
         private readonly IRepository<MessageReceived, Guid> _messageReceivedEventRepository;
         private readonly IRepository<Device, Guid> _deviceRepository;
         private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
+        private readonly IIoTDBProvider _dbProvider;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
@@ -48,7 +50,9 @@ namespace JiShe.CollectBus.Subscribers
             IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
             IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
             IRepository<MessageReceived, Guid> messageReceivedEventRepository,
-            IRepository<Device, Guid> deviceRepository, IMeterReadingRecordRepository meterReadingRecordsRepository)
+            IRepository<Device, Guid> deviceRepository,
+            IIoTDBProvider dbProvider,
+            IMeterReadingRecordRepository meterReadingRecordsRepository)
         {
             _logger = logger;
             _tcpService = tcpService;
@@ -58,6 +62,7 @@ namespace JiShe.CollectBus.Subscribers
             _messageReceivedEventRepository = messageReceivedEventRepository;
             _deviceRepository = deviceRepository;
             _meterReadingRecordsRepository = meterReadingRecordsRepository;
+            _dbProvider = dbProvider;
         }
 
         [CapSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
@@ -119,6 +124,8 @@ namespace JiShe.CollectBus.Subscribers
         [CapSubscribe(ProtocolConst.SubscriberReceivedEventName)]
         public async Task ReceivedEvent(MessageReceived receivedMessage)
         {
+            var currentTime = Clock.Now;
+
             var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
             if (protocolPlugin == null)
             {
@@ -126,6 +133,7 @@ namespace JiShe.CollectBus.Subscribers
             }
             else
             {
+                 
                 //todo 会根据不同的协议进行解析，然后做业务处理
                 TB3761 fN = await protocolPlugin.AnalyzeAsync<TB3761>(receivedMessage);
                 if(fN == null)
@@ -140,11 +148,8 @@ namespace JiShe.CollectBus.Subscribers
                     return;
                 }
 
-                //todo 查找是否有下发任务
-                 
-
-
-                await _meterReadingRecordsRepository.InsertAsync(new MeterReadingRecords()
+                //报文入库
+                var entity = new MeterReadingRecords()
                 {
                     ReceivedMessageHexString = receivedMessage.MessageHexString,
                     AFN = fN.Afn,
@@ -152,8 +157,18 @@ namespace JiShe.CollectBus.Subscribers
                     Pn = 0,
                     FocusAddress = "",
                     MeterAddress = "",
-                    //DataResult = tb3761FN.Text,
-                });
+                };
+
+                //如果没数据，则插入，有数据则更新
+                var updateEntity = await _meterReadingRecordsRepository.FirOrDefaultAsync(entity, currentTime);
+                if (updateEntity == null)
+                {
+                    await _meterReadingRecordsRepository.InsertAsync(entity, currentTime);
+                }
+
+
+                _dbProvider.InsertAsync();
+                //todo 查找是否有下发任务
 
                 //await _messageReceivedEventRepository.InsertAsync(receivedMessage);
             }

src/JiShe.CollectBus.Common/Consts/RedisConst.cs

@@ -43,6 +43,11 @@ namespace JiShe.CollectBus.Common.Consts
         /// </summary>
         public const string CacheTelemetryPacketInfoKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:TelemetryPacket:{"{2}"}:{"{3}"}:";
 
+        /// <summary>
+        /// 缓存设备平衡关系映射结果,{0}=>系统类型，{1}=>应用服务部署标记
+        /// </summary>
+        public const string CacheDeviceBalanceRelationMapResultKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:RelationMap";
+
         public const string CacheAmmeterFocusKey = "CacheAmmeterFocusKey";
     }
 }

src/JiShe.CollectBus.Common/Extensions/DateTimeExtensions.cs

@@ -175,7 +175,11 @@ namespace JiShe.CollectBus.Common.Extensions
         /// <returns></returns>
         public static string GetDataTableShardingStrategy(this DateTime dateTime)
         {
+#if DEBUG
             return $"{dateTime:yyyyMMddHHmm}";
+#else
+            return $"{dateTime:yyyyMMddHH}";
+#endif
         }
     }
 }

src/JiShe.CollectBus.Common/Helpers/DeviceGroupBalanceControl.cs

@@ -1,8 +1,10 @@
-using System;
+using JiShe.CollectBus.FreeRedisProvider;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
+using Volo.Abp.DependencyInjection;
 
 namespace JiShe.CollectBus.Common.Helpers
 {
@@ -12,25 +14,29 @@ namespace JiShe.CollectBus.Common.Helpers
     public class DeviceGroupBalanceControl
     { 
         /// <summary>
-        /// 设备组数量
+        /// 分组集合
         /// </summary>
-        private const int GroupCount = 50;
         private static List<string>[] _cachedGroups;
+
+        /// <summary>
+        /// 设备分组关系映射
+        /// </summary>
         private static Dictionary<string, int> _balancedMapping;
 
+
         /// <summary>
         /// 初始化缓存并强制均衡
         /// </summary>
-        public static void InitializeCache(List<string> deviceList)
+        public static void InitializeCache(List<string> deviceList,int groupCount = 50)
         {
             // 步骤1: 生成均衡映射表
-            _balancedMapping = CreateBalancedMapping(deviceList, GroupCount);
+            _balancedMapping = CreateBalancedMapping(deviceList, groupCount);
 
             // 步骤2: 根据映射表填充分组
-            _cachedGroups = new List<string>[GroupCount];
-            for (int i = 0; i < GroupCount; i++)
+            _cachedGroups = new List<string>[groupCount];
+            for (int i = 0; i < groupCount; i++)
             {
-                _cachedGroups[i] = new List<string>(capacity: deviceList.Count / GroupCount + 1);
+                _cachedGroups[i] = new List<string>(capacity: deviceList.Count / groupCount + 1);
             }
 
             foreach (var deviceId in deviceList)
@@ -41,7 +47,7 @@ namespace JiShe.CollectBus.Common.Helpers
         }
 
         /// <summary>
-        /// 通过 deviceId 获取分组
+        /// 通过 deviceId 获取所在的分组集合
         /// </summary>
         public static List<string> GetGroup(string deviceId)
         {
@@ -52,6 +58,17 @@ namespace JiShe.CollectBus.Common.Helpers
             return _cachedGroups[groupId];
         }
 
+        /// <summary>
+        /// 通过 deviceId 获取分组Id
+        /// </summary>
+        public static int GetDeviceGroupId(string deviceId)
+        {
+            if (_balancedMapping == null || _cachedGroups == null)
+                throw new InvalidOperationException("缓存未初始化");
+
+            return _balancedMapping[deviceId]; 
+        }
+
 
         /// <summary>
         /// 创建均衡映射表
@@ -69,9 +86,6 @@ namespace JiShe.CollectBus.Common.Helpers
             // 初始化分组计数器
             int[] groupCounters = new int[groupCount];
 
-            // 随机数生成器用于平衡分配
-            Random rand = new Random();
-
             foreach (var deviceId in deviceList)
             {
                 int preferredGroup = GetGroupId(deviceId, groupCount);

src/JiShe.CollectBus.Common/Helpers/JsonHelper.cs

@@ -30,6 +30,8 @@ namespace JiShe.CollectBus.Common.Helpers
                     DefaultIgnoreCondition = JsonIgnoreCondition.Never,
                     WriteIndented = false,
                     Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
+                    IgnoreReadOnlyFields = true,
+                    IgnoreReadOnlyProperties = true,
                 };
             }
 
@@ -67,6 +69,8 @@ namespace JiShe.CollectBus.Common.Helpers
                     DefaultIgnoreCondition = JsonIgnoreCondition.Never,
                     WriteIndented = false,
                     Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
+                    IgnoreReadOnlyFields = true,
+                    IgnoreReadOnlyProperties = true,
                 };
             }
 

src/JiShe.CollectBus.Common/JiShe.CollectBus.Common.csproj

@@ -25,6 +25,11 @@
 		<PackageReference Include="Serilog" Version="4.1.0" />
 		<PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
 		<PackageReference Include="Volo.Abp.Ddd.Domain" Version="8.3.3" />
+
+	</ItemGroup>
+
+	<ItemGroup>
+	  <ProjectReference Include="..\JiShe.CollectBus.FreeRedisProvider\JiShe.CollectBus.FreeRedisProvider.csproj" />
 	</ItemGroup>
 
 </Project>

src/JiShe.CollectBus.Domain/IotSystems/AFNEntity/AFNDataEntity.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.IotSystems.AFNEntity
+{
+    /// <summary>
+    /// AFN单项数据实体
+    /// </summary>
+    public class AFNDataEntity
+    {
+    }
+}

src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs

@@ -376,10 +376,9 @@ namespace JiShe.CollectBus.Host
 
             try
             { 
-                string serverTagName = configuration.GetSection(CommonConst.ServerTagName).Value!;
 
-                List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued(serverTagName);
-                topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived(serverTagName));
+                List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
+                topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
 
                 List<TopicSpecification> topicSpecifications = new List<TopicSpecification>();
                 foreach (var item in topics)

src/JiShe.CollectBus.Host/appsettings.json

@@ -35,12 +35,12 @@
   },
   "ConnectionStrings": {
     "Default": "mongodb://admin:admin02023@118.190.144.92:37117,118.190.144.92:37119,118.190.144.92:37120/JiSheCollectBus?authSource=admin&maxPoolSize=400&minPoolSize=10&waitQueueTimeoutMS=5000",
-    "Kafka": "121.42.242.91:29092,121.42.242.91:39092,121.42.242.91:49092",
+    "Kafka": "192.168.0.151:29092,192.168.0.151:39092,192.168.0.151:49092",
     "PrepayDB": "server=118.190.144.92;database=jishe.sysdb;uid=sa;pwd=admin@2023;Encrypt=False;Trust Server Certificate=False",
     "EnergyDB": "server=118.190.144.92;database=db_energy;uid=sa;pwd=admin@2023;Encrypt=False;Trust Server Certificate=False"
   },
   "Redis": {
-    "Configuration": "120.24.52.151:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
+    "Configuration": "192.168.0.151:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
     "DefaultDB": "14",
     "HangfireDB": "15"
   },
@@ -86,12 +86,13 @@
     "SecurityProtocol": "SASL_PLAINTEXT",
     "SaslMechanism": "PLAIN",
     "SaslUserName": "lixiao",
-    "SaslPassword": "lixiao1980"
+    "SaslPassword": "lixiao1980",
+    "NumPartitions": 50
   },
   "IoTDBOptions": {
     "UserName": "root",
     "Password": "root",
-    "ClusterList": [ "192.168.56.102:6667" ],
+    "ClusterList": [ "192.168.0.151:6667" ],
     "PoolSize": 2,
     "DataBaseName": "energy",
     "OpenDebugMode": true,

src/JiShe.CollectBus.MongoDB/Repository/MeterReadingRecord/MeterReadingRecordRepository.cs

@@ -103,8 +103,6 @@ namespace JiShe.CollectBus.Repository.MeterReadingRecord
         {
             var collection = await GetShardedCollection(entity.CreationTime);
 
-            var dbContext = await DbContextProvider.GetDbContextAsync();
-
             await collection.UpdateOneAsync(filter, update);
             return entity;
         }
@@ -120,8 +118,8 @@ namespace JiShe.CollectBus.Repository.MeterReadingRecord
         public async Task<MeterReadingRecords> FirOrDefaultAsync(MeterReadingRecords entity, DateTime? dateTime)
         {
             var collection = await GetShardedCollection(dateTime);
-            //await collection.findon
-            throw new NotImplementedException();
+            var query = await collection.FindAsync(d => d.CreationTime == dateTime.Value && d.AFN == entity.AFN && d.Fn == entity.Fn && d.FocusAddress == entity.FocusAddress);
+            return await query.FirstOrDefaultAsync();
         }
 
         /// <summary>

src/JiShe.CollectBus.Protocol.Contracts/Extensions/ProtocolConstExtensions.cs

@@ -15,7 +15,7 @@ namespace JiShe.CollectBus.Protocol.Contracts
         /// 自动获取 ProtocolConst 类中所有下行 Kafka 主题名称
         /// （通过反射筛选 public const string 且字段名以 "EventName" 结尾的常量）
         /// </summary>
-        public static List<string> GetAllTopicNamesByIssued(string serverTagName)
+        public static List<string> GetAllTopicNamesByIssued()
         {
             List<string> topics = typeof(ProtocolConst)
                 .GetFields(BindingFlags.Public | BindingFlags.Static)
@@ -24,7 +24,6 @@ namespace JiShe.CollectBus.Protocol.Contracts
                     !f.IsInitOnly &&
                     f.FieldType == typeof(string) &&
                     f.Name.EndsWith("IssuedEventName")) // 通过命名规则过滤主题字段
-                //.Select(f => $"{serverTagName}.{(string)f.GetRawConstantValue()!}")
                 .Select(f => (string)f.GetRawConstantValue()!)
                 .ToList();
 
@@ -35,7 +34,7 @@ namespace JiShe.CollectBus.Protocol.Contracts
         /// 自动获取 ProtocolConst 类中所有下行 Kafka 主题名称
         /// （通过反射筛选 public const string 且字段名以 "EventName" 结尾的常量）
         /// </summary>
-        public static List<string> GetAllTopicNamesByReceived(string serverTagName)
+        public static List<string> GetAllTopicNamesByReceived()
         {
             //固定的上报主题
             var topicList = typeof(ProtocolConst)
@@ -50,13 +49,20 @@ namespace JiShe.CollectBus.Protocol.Contracts
 
             //动态上报主题，需根据协议的AFN功能码动态获取
             var afnList = EnumExtensions.ToNameValueDictionary<AFN>();
+
+            //需要排除的AFN功能码
+            var excludeItems = new List<int>() { 6, 7, 8,15 };
+
             foreach (var item in afnList)
             {
+                if (excludeItems.Contains(item.Value))
+                {
+                    continue;
+                }
+
                 topicList.Add(string.Format(ProtocolConst.AFNTopicNameFormat, item.Value.ToString().PadLeft(2, '0')));
             }
 
-            //return topicList.Select(f => $"{serverTagName}.{f}").ToList();
-
             return topicList;
         }
     }

src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs

@@ -19,7 +19,7 @@ namespace JiShe.CollectBus.Protocol.Contracts
         public const string SubscriberLoginIssuedEventName = "issued.login.event";
 
         /// <summary>
-        /// 上行消息主题
+        /// 上行消息主题,测试使用
         /// </summary>
         public const string SubscriberReceivedEventName = "received.event";
 
@@ -106,7 +106,71 @@ namespace JiShe.CollectBus.Protocol.Contracts
         /// <summary>
         /// AFN上行主题格式
         /// </summary>
-        public const string AFNTopicNameFormat = "received.afn{0}.event";
+        public const string AFNTopicNameFormat = "received.afn{0}h.event";
 
+        /// <summary>
+        /// AFN00H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN00ReceivedEventNameTemp = "received.afn00h.event";
+
+        /// <summary>
+        /// AFN01H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN00HReceivedEventNameTemp = "received.afn01h.event";
+
+        /// <summary>
+        /// AFN02H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN01HReceivedEventNameTemp = "received.afn02h.event";
+
+        /// <summary>
+        /// AFN03H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN02HReceivedEventNameTemp = "received.afn03h.event";
+
+        /// <summary>
+        /// AFN04H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN04HReceivedEventNameTemp = "received.afn04h.event";
+
+        /// <summary>
+        /// AFN05H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN05HReceivedEventNameTemp = "received.afn05h.event";
+
+        /// <summary>
+        /// AFN09H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN09HReceivedEventNameTemp = "received.afn09h.event";
+
+        /// <summary>
+        /// AFN0AH上行主题格式
+        /// </summary>
+        public const string SubscriberAFN0AHReceivedEventNameTemp = "received.afn10h.event";
+
+        /// <summary>
+        /// AFN0BH上行主题格式
+        /// </summary>
+        public const string SubscriberAFN0BHReceivedEventNameTemp = "received.afn11h.event";
+
+        /// <summary>
+        /// AFN0CH上行主题格式
+        /// </summary>
+        public const string SubscriberAFN0CHReceivedEventNameTemp = "received.afn12h.event";
+
+        /// <summary>
+        /// AFN0DH上行主题格式
+        /// </summary>
+        public const string SubscriberAFN0DHReceivedEventNameTemp = "received.afn13h.event";
+
+        /// <summary>
+        /// AFN0EH上行主题格式
+        /// </summary>
+        public const string SubscriberAFN0EHReceivedEventNameTemp = "received.afn14h.event";
+
+        /// <summary>
+        /// AFN10H上行主题格式
+        /// </summary>
+        public const string SubscriberAFN10HReceivedEventNameTemp = "received.afn16h.event";
     }
 }