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

src/JiShe.CollectBus.Application.Contracts/JiShe.CollectBus.Application.Contracts.csproj

@@ -22,4 +22,8 @@
     <ProjectReference Include="..\JiShe.CollectBus.Domain\JiShe.CollectBus.Domain.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <Folder Include="Workers\" />
+  </ItemGroup>
+
 </Project>

src/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs

@@ -1,4 +1,5 @@
 using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.GatherItem;
 using JiShe.CollectBus.Watermeter;
 using System;
 using System.Collections.Generic;
@@ -16,6 +17,12 @@ namespace JiShe.CollectBus.Workers
     public interface IScheduledMeterReadingService : IApplicationService
     {
 
+        /// <summary>
+        /// 获取采集项列表
+        /// </summary>
+        /// <returns></returns>
+        Task<List<GatherItemInfo>> GetGatherItemByDataTypes();
+
         #region 电表采集处理
         /// <summary>
         /// 获取电表信息

src/JiShe.CollectBus.Application/Consumers/IssuedConsumer.cs

@@ -12,7 +12,7 @@ namespace JiShe.CollectBus.Consumers
 {
     public class IssuedConsumer: IConsumer<MessageIssued>
     {
-        private readonly ILogger<ReceivedHeartbeatConsumer> _logger;
+        private readonly ILogger<IssuedConsumer> _logger;
         private readonly ITcpService _tcpService;
         private readonly IRepository<MessageReceivedLogin, Guid> _messageReceivedLoginEventRepository;
         private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
@@ -24,7 +24,7 @@ namespace JiShe.CollectBus.Consumers
         /// <param name="tcpService"></param>
         /// <param name="messageReceivedLoginEventRepository"></param>
         /// <param name="messageReceivedHeartbeatEventRepository"></param>
-        public IssuedConsumer(ILogger<ReceivedHeartbeatConsumer> logger, 
+        public IssuedConsumer(ILogger<IssuedConsumer> logger, 
             ITcpService tcpService, 
             IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository, 
             IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository)

src/JiShe.CollectBus.Application/Consumers/WorkerConsumer.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Threading.Tasks;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.Common.Models;
+using JiShe.CollectBus.MessageIssueds;
+using JiShe.CollectBus.MessageReceiveds;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using TouchSocket.Sockets;
+using Volo.Abp.Domain.Repositories;
+
+namespace JiShe.CollectBus.Consumers
+{
+    /// <summary>
+    /// 定时抄读任务消费者
+    /// </summary>
+    public class WorkerConsumer : IConsumer<IssuedEventMessage>
+    {
+        private readonly ILogger<WorkerConsumer> _logger;
+        private readonly ITcpService _tcpService;
+
+        /// <summary>
+        /// WorkerConsumer
+        /// </summary>
+        /// <param name="logger"></param>
+        /// <param name="tcpService"></param>
+        public WorkerConsumer(ILogger<WorkerConsumer> logger, 
+            ITcpService tcpService)
+        {
+            _logger = logger;
+            _tcpService = tcpService;
+        }
+
+
+        public async Task Consume(ConsumeContext<IssuedEventMessage> context)
+        {
+            await _tcpService.SendAsync(context.Message.ClientId, context.Message.Message);
+        }
+    }
+}

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

@@ -0,0 +1,672 @@
+using DotNetCore.CAP;
+using FreeRedis;
+using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.Common.BuildSendDatas;
+using JiShe.CollectBus.Common.Consts;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.Common.Helpers;
+using JiShe.CollectBus.GatherItem;
+using JiShe.CollectBus.MessageReceiveds;
+using JiShe.CollectBus.Protocol.Contracts;
+using JiShe.CollectBus.Watermeter;
+using MassTransit;
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+using System.Linq;
+using System.Net;
+using System.Text;
+using System.Threading.Tasks;
+using TouchSocket.Sockets;
+using Volo.Abp.Application.Services;
+using static FreeSql.Internal.GlobalFilter;
+
+namespace JiShe.CollectBus.Workers
+{
+    /// <summary>
+    /// 定时采集服务
+    /// </summary>
+    public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
+    {
+        private readonly ILogger<BasicScheduledMeterReadingService> _logger;
+        private readonly ICapPublisher _capBus;
+        public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger, ICapPublisher capBus)
+        {
+            _capBus = capBus;
+            _logger = logger;
+        }
+
+        /// <summary>
+        /// 系统类型
+        /// </summary>
+        public abstract string SystemType { get; }
+
+        /// <summary>
+        ///电表日冻结采集项
+        /// </summary>
+        protected List<string> DayFreezeCodes = new List<string>() { "0D_3", "0D_4", "0D_161", "0D_162", "0D_163", "0D_164", "0D_165", "0D_166", "0D_167", "0D_168", "0C_149", };
+
+        /// <summary>
+        /// 电表月冻结采集项
+        /// </summary>
+        protected List<string> MonthFreezeCodes = new List<string>() { "0D_177", "0D_178", "0D_179", "0D_180", "0D_181", "0D_182", "0D_183", "0D_184", "0D_193", "0D_195", };
+
+        /// <summary>
+        /// 获取采集项列表
+        /// </summary>
+        /// <returns></returns>
+        public virtual Task<List<GatherItemInfo>> GetGatherItemByDataTypes()
+        {
+            throw new NotImplementedException($"{nameof(GetGatherItemByDataTypes)}请根据不同系统类型进行实现");
+        }
+
+        #region 电表采集处理
+
+        /// <summary>
+        /// 获取电表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        public virtual Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "")
+        {
+            throw new NotImplementedException($"{nameof(GetAmmeterInfoList)}请根据不同系统类型进行实现");
+        }
+
+        /// <summary>
+        /// 初始化电表缓存数据
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        public virtual async Task InitAmmeterCacheData(string gatherCode = "")
+        {
+            var meterInfos = await GetAmmeterInfoList(gatherCode);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，电表数据为空");
+            }
+
+            //获取采集项类型数据
+            var gatherItemInfos = await GetGatherItemByDataTypes();
+            if (gatherItemInfos == null || gatherItemInfos.Count <= 0)
+            {
+                throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，采集项类型数据为空");
+            }
+
+            //根据采集频率分组，获得采集频率分组
+            var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
+            foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
+            {
+                //将表计信息根据集中器分组，获得集中器号
+                var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList();
+                foreach (var item in meterInfoGroup)
+                {
+                    if (string.IsNullOrWhiteSpace(item.Key))
+                    {
+                        continue;
+                    }
+
+                    var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}";
+                    Dictionary<string, AmmeterInfo> keyValuePairs = new Dictionary<string, AmmeterInfo>();
+                    foreach (var ammeter in item)
+                    {
+                        //处理ItemCode
+                        if (string.IsNullOrWhiteSpace(ammeter.ItemCodes))
+                        {
+                            var itemArr = ammeter.DataTypes.Split(',').ToList();
+
+                            #region 拼接采集项
+                            List<string> itemCodeList = new List<string>();
+                            foreach (var dataType in itemArr)
+                            {
+                                var excludeItemCode = "10_98,10_94";//排除透明转发:尖峰平谷时段、跳合闸
+                                var gatherItem = gatherItemInfos.FirstOrDefault(f => f.DataType.Equals(dataType));
+                                if (gatherItem != null)
+                                {
+                                    if (!excludeItemCode.Contains(gatherItem.ItemCode))
+                                    {
+                                        itemCodeList.Add(gatherItem.ItemCode);
+                                    }
+                                }
+                            }
+                            #endregion
+
+                            #region 特殊电表采集项编号处理
+                            if (itemArr.Exists(e => e.Equals("95"))) //德力西DTS
+                            {
+                                itemCodeList.Add("10_95");
+                            }
+                            //if (itemArr.Exists(e => e.Equals("109")))//WAVE_109
+                            //    ammeter.ItemCodes += "10_109,";
+                            #endregion
+
+                            ammeter.ItemCodes = itemCodeList.Serialize();//转换成JSON字符串
+
+                            if (!string.IsNullOrWhiteSpace(ammeter.ItemCodes))
+                            {
+                                ammeter.ItemCodes = ammeter.ItemCodes.Replace("WAVE_109", "10_109");
+                            }
+                        }
+
+                        keyValuePairs.TryAdd($"{ammeter.ID}", ammeter);
+                    }
+                    await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs);
+                }
+            }
+
+            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成");
+        }
+
+        /// <summary>
+        /// 1分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task AmmeterScheduledMeterOneMinuteReading()
+        {
+            //获取缓存中的电表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, AmmeterInfo>> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList, 1);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+
+            await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, meterInfos);
+
+            _logger.LogInformation($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理完成");
+
+        }
+
+        /// <summary>
+        /// 5分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task AmmeterScheduledMeterFiveMinuteReading()
+        {
+            //获取缓存中的电表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 5)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, AmmeterInfo>> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList, 5);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+            await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, meterInfos);
+
+            _logger.LogInformation($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理完成");
+        }
+
+        /// <summary>
+        /// 15分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task AmmeterScheduledMeterFifteenMinuteReading()
+        {
+            //获取缓存中的电表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 15)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, AmmeterInfo>> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList, 15);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+
+            await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, meterInfos);
+            _logger.LogInformation($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理完成");
+        }
+        #endregion
+
+
+        #region 水表采集处理
+
+        /// <summary>
+        /// 获取水表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        public virtual Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "")
+        {
+            throw new NotImplementedException($"{nameof(GetWatermeterInfoList)}请根据不同系统类型进行实现");
+        }
+
+        /// <summary>
+        /// 初始化水表缓存数据
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        public virtual async Task InitWatermeterCacheData(string gatherCode = "")
+        {
+            var meterInfos = await GetWatermeterInfoList(gatherCode);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化水表缓存数据时，水表数据为空");
+            }
+
+            //获取采集项类型数据
+            var gatherItemInfos = await GetGatherItemByDataTypes();
+            if (gatherItemInfos == null || gatherItemInfos.Count <= 0)
+            {
+                throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化水表缓存数据时，采集项类型数据为空");
+            }
+
+            //根据采集频率分组，获得采集频率分组
+            var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
+            foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
+            {
+                //将表计信息根据集中器分组，获得集中器号
+                var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList();
+                foreach (var item in meterInfoGroup)
+                {
+                    if (string.IsNullOrWhiteSpace(item.Key))
+                    {
+                        continue;
+                    }
+
+                    var redisCacheKey = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}";
+                    Dictionary<string, WatermeterInfo> keyValuePairs = new Dictionary<string, WatermeterInfo>();
+                    foreach (var subItem in item)
+                    {
+
+                        keyValuePairs.TryAdd($"{subItem.ID}", subItem);
+                    }
+                    await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs);
+                }
+            }
+            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化水表缓存数据完成");
+        }
+
+        /// <summary>
+        /// 1分钟采集水表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task WatermeterScheduledMeterOneMinuteReading()
+        {
+            //获取缓存中的水表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 1)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, WatermeterInfo>> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList, 1);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+
+            _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成");
+        }
+
+        /// <summary>
+        /// 5分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task WatermeterScheduledMeterFiveMinuteReading()
+        {
+
+            //获取缓存中的水表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 5)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, WatermeterInfo>> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList, 5);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+
+            _logger.LogInformation($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理完成");
+        }
+
+        /// <summary>
+        /// 15分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual async Task WatermeterScheduledMeterFifteenMinuteReading()
+        {
+            //获取缓存中的水表信息
+            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 15)}*";
+            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
+            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表据处理时没有获取到缓存信息,-101");
+                return;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            Dictionary<string, Dictionary<string, WatermeterInfo>> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList, 15);
+            if (meterInfos == null || meterInfos.Count <= 0)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理时没有获取到缓存信息,-102");
+                return;
+            }
+
+            _logger.LogInformation($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理完成");
+        }
+        #endregion
+
+
+        #region 公共处理方法
+        /// <summary>
+        /// 批量获取缓存的表计信息
+        /// </summary>
+        /// <typeparam name="T">表信息数据对象</typeparam>
+        /// <param name="redisKeys">采集频率对应的缓存Key集合</param>
+        /// <param name="minute">采集频率，1分钟、5分钟、15分钟</param>
+        /// <returns></returns>
+        private async Task<Dictionary<string, Dictionary<string, T>>> GetMeterCacheData<T>(string[] redisKeys, int minute)
+        {
+            //通过lua脚本一次性获取所有缓存内容
+            var luaScript = @"
+                local results = {}
+                for i, key in ipairs(KEYS) do
+                    local data = redis.call('HGETALL', key)
+                    results[i] = {key, data}
+                end
+                return results";
+            var oneMinuteAmmerterResult = await FreeRedisProvider.FreeRedis.EvalAsync(luaScript, redisKeys); //传递 KEYS
+            if (oneMinuteAmmerterResult == null)
+            {
+                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 定时任务采集表数据处理时没有获取到缓存信息,-102");
+                return null;
+            }
+
+            // 解析结果（结果为嵌套数组）
+            var meterInfos = new Dictionary<string, Dictionary<string, T>>(); ;
+            if (oneMinuteAmmerterResult is object[] arr)
+            {
+                foreach (object[] item in arr)
+                {
+                    string key = (string)item[0];//集中器地址对应的Redis缓存Key
+                    object[] fieldsAndValues = (object[])item[1];//缓存Key对应的Hash表数据集合
+                    var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, minute)}";
+                    string focusAddress = key.Replace(redisCacheKey, "");//集中器地址
+
+                    var meterHashs = new Dictionary<string, T>();
+                    for (int i = 0; i < fieldsAndValues.Length; i += 2)
+                    {
+                        string meterld = (string)fieldsAndValues[i];//表ID
+                        string meterStr = (string)fieldsAndValues[i + 1];//表详情数据
+
+                        T meterInfo = default;
+                        if (!string.IsNullOrWhiteSpace(meterStr))
+                        {
+                            meterInfo = meterStr.Deserialize<T>()!;
+                        }
+                        if (meterInfo != null)
+                        {
+                            meterHashs[meterld] = meterInfo;
+                        }
+                        else
+                        {
+                            _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 定时任务采集表数据处理时集中器缓存{key}数据的{meterld}处理异常");
+                        }
+                    }
+                    meterInfos[focusAddress] = meterHashs;
+                }
+            }
+
+            return meterInfos;
+        }
+
+        /// <summary>
+        /// 电表采集任务指令创建
+        /// </summary>
+        /// <param name="eventName">采集频率订阅主题</param>
+        /// <param name="focusGroup">集中器数据分组</param>
+        /// <returns></returns>
+        private async Task AmmerterScheduledMeterReadingIssued(string eventName, Dictionary<string, Dictionary<string, AmmeterInfo>> focusGroup)
+        {
+            if (string.IsNullOrWhiteSpace(eventName) || focusGroup == null || focusGroup.Count <= 0)
+            {
+                _logger.LogError($"{nameof(AmmerterScheduledMeterReadingIssued)} 电表数据采集指令生成失败，参数异常,-101");
+                return;
+            }
+            try
+            {
+                //将采集器编号的hash值取模分组
+                const int TotalShards = 20;
+                var focusHashGroups = new Dictionary<int, Dictionary<string, Dictionary<string, AmmeterInfo>>>();
+
+                foreach (var (collectorId, ammetersDictionary) in focusGroup)
+                {
+                    if (string.IsNullOrWhiteSpace(collectorId))
+                    {
+                        _logger.LogError($"{nameof(AmmerterScheduledMeterReadingIssued)} 集中器信息分组取模失败，无效Key -102");
+                        continue;
+                    }
+
+                    // 计算哈希分组ID
+                    int hashGroupId = Math.Abs(collectorId.GetHashCode() % TotalShards);
+
+                    // 获取或创建分组（避免重复查找）
+                    if (!focusHashGroups.TryGetValue(hashGroupId, out var group))
+                    {
+                        group = new Dictionary<string, Dictionary<string, AmmeterInfo>>();
+                        focusHashGroups[hashGroupId] = group;
+                    }
+
+                    // 将当前集中器数据加入分组
+                    group[collectorId] = ammetersDictionary;
+                }
+
+                if (focusHashGroups == null)
+                {
+                    _logger.LogError($"{nameof(AmmerterScheduledMeterReadingIssued)} 集中器信息分组取模失败 -103");
+                    return;
+                }
+
+                //根据分组创建线程批处理集中器
+                foreach (var group in focusHashGroups)
+                {
+                    _= Task.Run(async () => { await CreatePublishTask(eventName,group.Value); });
+                }
+
+                await Task.CompletedTask;
+            }
+            catch (Exception)
+            {
+
+                throw;
+            } 
+        }
+
+        /// <summary>
+        /// 创建发布任务
+        /// </summary>
+        /// <param name="eventName"></param>
+        /// <param name="focusGroup"></param>
+        /// <returns></returns>
+        private async Task CreatePublishTask(string eventName, Dictionary<string, Dictionary<string, AmmeterInfo>> focusGroup)
+        {
+            foreach (var focusInfo in focusGroup)
+            {
+                foreach (var ammeterInfo in focusInfo.Value)
+                { 
+                    var meter = ammeterInfo.Value;
+
+                    if (string.IsNullOrWhiteSpace(meter.ItemCodes))
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 集中器{meter.FocusAddress}的电表{meter.Name}数据采集指令生成失败，采集项为空,-101");
+                        continue;
+                    }
+
+                    //载波的不处理
+                    if (meter.MeteringPort == (int)MeterLinkProtocolEnum.Carrierwave)
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 集中器{meter.FocusAddress}的电表{meter.Name}数据采集指令生成失败，载波不处理,-102");
+                        continue;
+                    }
+
+                    if (meter.State.Equals(2))
+                    {
+                        _logger.LogWarning($"{nameof(CreatePublishTask)} {meter.Name} 集中器{meter.FocusAddress}的电表{meter.Name}状态为禁用，不处理");
+                        continue;
+                    }
+
+                    //排除1天未在线的集中器生成指令 或 排除集中器配置为自动上报的集中器
+                    if (!IsGennerateCmd(meter.LastTime, -1))
+                    {
+                        _logger.LogInformation($"{nameof(CreatePublishTask)} 集中器{meter.FocusAddress}的电表{meter.Name}，采集时间：{meter.LastTime}，已超过1天未在线，不生成指令");
+                        continue;
+                    }
+
+                    if (string.IsNullOrWhiteSpace(meter.AreaCode))
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 表ID:{meter.ID},集中器通信区号为空");
+                        continue;
+                    }
+                    if (string.IsNullOrWhiteSpace(meter.Address))
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 表ID:{meter.ID},集中器通信地址为空");
+                        continue;
+                    }
+                    if (Convert.ToInt32(meter.Address) > 65535)
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 表ID:{meter.ID},集中器通信地址无效,确保大于65535");
+                        continue;
+                    }
+                    if (meter.MeteringCode <= 0 || meter.MeteringCode > 2033)
+                    {
+                        _logger.LogError($"{nameof(CreatePublishTask)} 表ID:{meter.ID},非有效测量点号({meter.MeteringCode})");
+                        continue;
+                    }
+
+                    List<string> tempCodes = meter.ItemCodes.Deserialize<List<string>>()!;
+
+                    //TODO:自动上报数据只主动采集1类数据。
+                    if (meter.AutomaticReport.Equals(1))
+                    {
+                        var tempSubCodes = new List<string>();
+                        var tempItemCodes = string.Empty;
+                        if (meter.ItemCodes.Contains("0C_49"))
+                            tempItemCodes += "0C_49,";
+                        if (meter.ItemCodes.Contains("0C_149"))
+                            tempItemCodes += "0C_149,";
+                        if (meter.ItemCodes.Contains("10_97"))
+                            tempItemCodes += "10_97";
+
+                        if (string.IsNullOrWhiteSpace(tempItemCodes))
+                        {
+                            continue;
+                        }
+                        else
+                        {
+                            meter.ItemCodes = tempItemCodes;
+                        }
+                    }
+
+                    foreach (var tempItem in tempCodes)
+                    {
+                        //排除已发送日冻结和月冻结采集项配置
+                        if(DayFreezeCodes.Contains(tempItem))
+                        {
+                            continue;
+                        }
+
+                        if (MonthFreezeCodes.Contains(tempItem))
+                        {
+                            continue;
+                        }
+
+                    }
+                    //排除已发送日冻结和月冻结采集项配置
+                    //if (!isSendDayFreeze)
+                    meter.ItemCodes = meter.ItemCodes.Replace("0D_3", "").Replace("0D_4", "")
+                        .Replace("0D_161", "").Replace("0D_162", "").Replace("0D_163", "").Replace("0D_164", "")
+                        .Replace("0D_165", "").Replace("0D_166", "").Replace("0D_167", "").Replace("0D_168", "").Replace("0C_149", "");
+
+                    //if (!isSendMonthFreeze)
+                    meter.ItemCodes = meter.ItemCodes.Replace("0D_177", "").Replace("0D_178", "").Replace("0D_179", "").Replace("0D_180", "")
+                        .Replace("0D_181", "").Replace("0D_181", "").Replace("0D_182", "").Replace("0D_183", "").Replace("0D_184", "")
+                        .Replace("0D_193", "").Replace("0D_195", "");
+
+                   
+
+                    //TODO:特殊表
+
+
+                    //var itemCodeArr = itemCode.Split('_');
+                    //var aFN = (AFN)itemCodeArr[0].HexToDec();
+                    //var fn = int.Parse(itemCodeArr[1]);
+                    //if (aFN == AFN.请求实时数据)
+                    //{
+                    //    var bytes = Build3761SendData.BuildAmmeterReadRealTimeDataSendCmd(address, ammeter.MeterCode.Value, (ATypeOfDataItems)fn);
+                    //    bytesList.Add(bytes);
+                    //}
+                    //else if (aFN == AFN.请求历史数据)
+                    //{
+                    //    var density = (FreezeDensity)input.Density;
+                    //    var bytes = Build3761SendData.BuildAmmeterReadingIIdataTypeItemsSendCmd(address, ammeter.MeterCode.Value, (IIdataTypeItems)fn, density, 0);
+                    //    bytesList.Add(bytes);
+                    //}
+
+                }
+            }
+
+            string deviceNo = "";
+            string messageHexString = "";
+
+            var messageReceivedHeartbeatEvent = new MessageReceivedHeartbeat
+            {
+                //ClientId = client.Id,
+                //ClientIp = client.IP,
+                //ClientPort = client.Port,
+                MessageHexString = messageHexString,
+                DeviceNo = deviceNo,
+                MessageId = NewId.NextGuid().ToString()
+            };
+            await _capBus.PublishAsync(eventName, messageReceivedHeartbeatEvent);
+        }
+
+        /// <summary>
+        /// 指定时间对比当前时间
+        /// </summary>
+        /// <param name="lastTime"></param>
+        /// <param name="subtrahend"></param>
+        /// <returns></returns>
+        private bool IsGennerateCmd(DateTime lastTime, int subtrahend = 0)
+        {
+            if (DateTime.Now.AddDays(subtrahend) >= lastTime)//当前时间减去一天，大于等于最后在线时间，不再生成该集中器下表生成采集指令
+                return false;
+            return true;
+        }
+
+        #endregion
+    }
+}

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

@@ -1,10 +1,12 @@
 using DotNetCore.CAP;
 using FreeRedis;
+using FreeSql;
 using JiShe.CollectBus.Ammeters;
 using JiShe.CollectBus.Common.Consts;
 using JiShe.CollectBus.Devices;
 using JiShe.CollectBus.FreeRedisProvider;
 using JiShe.CollectBus.FreeSql;
+using JiShe.CollectBus.GatherItem;
 using JiShe.CollectBus.Watermeter;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
@@ -34,6 +36,25 @@ namespace JiShe.CollectBus.Workers
 
         public sealed override string SystemType => SystemTypeConst.Energy;
 
+        /// <summary>
+        /// 获取采集项列表
+        /// </summary>
+        /// <returns></returns>
+        public override async Task<List<GatherItemInfo>> GetGatherItemByDataTypes()
+        {
+            try
+            {
+                string sql = $"SELECT DataType,ItemCode FROM TB_GatherItem(NOLOCK) WHERE [State]=0";
+                return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
+               .Ado
+               .QueryAsync<GatherItemInfo>(sql, null);
+            }
+            catch
+            {
+                return null;
+            }
+        }
+
         /// <summary>
         /// 获取电表信息
         /// </summary>

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

@@ -1,358 +0,0 @@
-using DotNetCore.CAP;
-using FreeRedis;
-using JiShe.CollectBus.Ammeters;
-using JiShe.CollectBus.Common.Consts;
-using JiShe.CollectBus.Common.Helpers;
-using JiShe.CollectBus.Watermeter;
-using Microsoft.Extensions.Logging;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using Volo.Abp.Application.Services;
-using static FreeSql.Internal.GlobalFilter;
-
-namespace JiShe.CollectBus.Workers
-{
-    /// <summary>
-    /// 定时采集服务
-    /// </summary>
-    public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
-    {
-        private readonly ILogger<BasicScheduledMeterReadingService> _logger;
-        private readonly ICapPublisher _capBus;
-        public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger, ICapPublisher capBus)
-        {
-            _capBus = capBus;
-            _logger = logger;
-        }
-
-        /// <summary>
-        /// 系统类型
-        /// </summary>
-        public abstract string SystemType { get; }
-
-        #region 电表采集处理
-
-        /// <summary>
-        /// 获取电表信息
-        /// </summary>
-        /// <param name="gatherCode">采集端Code</param>
-        /// <returns></returns>
-        public virtual Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "")
-        {
-            throw new NotImplementedException($"{nameof(GetAmmeterInfoList)}请根据不同系统类型进行实现");
-        }
-
-        /// <summary>
-        /// 初始化电表缓存数据
-        /// </summary>
-        /// <param name="gatherCode">采集端Code</param>
-        /// <returns></returns>
-        public virtual async Task InitAmmeterCacheData(string gatherCode = "")
-        {
-            var meterInfos = await GetAmmeterInfoList(gatherCode);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化电表缓存数据时，电表数据为空");
-            }
-
-            //根据采集频率分组
-            var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
-            foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
-            {
-                //将表计信息根据集中器分组
-                var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList();
-                foreach (var item in meterInfoGroup)
-                {
-                    if (string.IsNullOrWhiteSpace(item.Key))
-                    {
-                        continue;
-                    }
-
-                    var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}";
-                    Dictionary<string, AmmeterInfo> keyValuePairs = new Dictionary<string, AmmeterInfo>();
-                    foreach (var subItem in item)
-                    {
-
-                        keyValuePairs.TryAdd($"{subItem.ID}", subItem);
-                    }
-                    await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs);
-                }
-            }
-
-            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成");
-        }
-
-        /// <summary>
-        /// 1分钟采集电表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task AmmeterScheduledMeterOneMinuteReading()
-        {
-            //获取缓存中的电表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-101");
-                return;
-            }
-              
-            // 解析结果（结果为嵌套数组）
-            List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理完成");
-
-        }
-
-        /// <summary>
-        /// 5分钟采集电表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task AmmeterScheduledMeterFiveMinuteReading()
-        {
-            //获取缓存中的电表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 5)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-101");
-                return;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理完成");
-        }
-
-        /// <summary>
-        /// 15分钟采集电表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task AmmeterScheduledMeterFifteenMinuteReading()
-        {
-            //获取缓存中的电表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 15)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理时没有获取到缓存信息,-101");
-                return;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理完成");
-        }
-        #endregion
-
-
-        #region 水表采集处理
-
-        /// <summary>
-        /// 获取水表信息
-        /// </summary>
-        /// <param name="gatherCode">采集端Code</param>
-        /// <returns></returns>
-        public virtual Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "")
-        {
-            throw new NotImplementedException($"{nameof(GetWatermeterInfoList)}请根据不同系统类型进行实现");
-        }
-
-        /// <summary>
-        /// 初始化水表缓存数据
-        /// </summary>
-        /// <param name="gatherCode">采集端Code</param>
-        /// <returns></returns>
-        public virtual async Task InitWatermeterCacheData(string gatherCode = "")
-        {
-            var meterInfos = await GetWatermeterInfoList(gatherCode);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化水表缓存数据时，水表数据为空");
-            }
-            //根据采集频率分组
-            var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
-            foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
-            {
-                //将表计信息根据集中器分组
-                var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList();
-                foreach (var item in meterInfoGroup)
-                {
-                    if (string.IsNullOrWhiteSpace(item.Key))
-                    {
-                        continue;
-                    }
-
-                    var redisCacheKey = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}";
-                    Dictionary<string, WatermeterInfo> keyValuePairs = new Dictionary<string, WatermeterInfo>();
-                    foreach (var subItem in item)
-                    {
-
-                        keyValuePairs.TryAdd($"{subItem.ID}", subItem);
-                    }
-                    await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs);
-                }
-            }
-            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化水表缓存数据完成");
-        }
-
-        /// <summary>
-        /// 1分钟采集水表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task WatermeterScheduledMeterOneMinuteReading()
-        {
-            //获取缓存中的水表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 1)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表据处理时没有获取到缓存信息,-101");
-                return;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成");
-        }
-
-        /// <summary>
-        /// 5分钟采集电表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task WatermeterScheduledMeterFiveMinuteReading()
-        {
-
-            //获取缓存中的水表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 5)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表据处理时没有获取到缓存信息,-101");
-                return;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理完成");
-        }
-
-        /// <summary>
-        /// 15分钟采集电表数据
-        /// </summary>
-        /// <returns></returns>
-        public virtual async Task WatermeterScheduledMeterFifteenMinuteReading()
-        {
-            //获取缓存中的水表信息
-            var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 15)}*";
-            var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
-            if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表据处理时没有获取到缓存信息,-101");
-                return;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
-            if (meterInfos == null || meterInfos.Count <= 0)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理时没有获取到缓存信息,-102");
-                return;
-            }
-
-            _logger.LogInformation($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理完成");
-        }
-        #endregion
-
-
-        #region 公共处理方法
-        /// <summary>
-        /// 批量获取缓存的表计信息
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisKeys"></param>
-        /// <returns></returns>
-        private async Task<List<T>> GetMeterCacheData<T>(string[] redisKeys)
-        {
-            //通过lua脚本一次性获取所有缓存内容
-            var luaScript = @"
-                local results = {}
-                for i, key in ipairs(KEYS) do
-                    local data = redis.call('HGETALL', key)
-                    results[i] = {key, data}
-                end
-                return results";
-            var oneMinuteAmmerterResult = await FreeRedisProvider.FreeRedis.EvalAsync(luaScript, redisKeys); //传递 KEYS
-            if (oneMinuteAmmerterResult == null)
-            {
-                _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102");
-                return null;
-            }
-
-            // 解析结果（结果为嵌套数组）
-            List<T> meterInfos = new List<T>();
-            if (oneMinuteAmmerterResult is object[] arr)
-            {
-                foreach (object[] item in arr)
-                {
-                    string key = (string)item[0];
-                    object[] fieldsAndValues = (object[])item[1];
-
-                    for (int i = 0; i < fieldsAndValues.Length; i += 2)
-                    {
-                        string field = (string)fieldsAndValues[i];
-                        string valueStr = (string)fieldsAndValues[i + 1];
-                        T value = default;
-                        if (!string.IsNullOrWhiteSpace(valueStr))
-                        {
-                            value = valueStr.Deserialize<T>()!;
-                        }
-                        if (value != null)
-                        {
-                            meterInfos.Add(value);
-                        }
-                        else
-                        {
-                            _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表{key}数据{field}处理异常");
-                        }
-                    }
-                }
-            }
-
-            return meterInfos;
-        }
-        #endregion
-    }
-}

src/JiShe.CollectBus.Common/Enums/MeterLinkProtocolEnum.cs

@@ -0,0 +1,54 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Enums
+{
+    /// <summary>
+    /// 表计连接通讯协议--表计与集中器的通讯协议    
+    /// </summary>
+    public enum MeterLinkProtocolEnum
+    {
+        /// <summary>
+        /// 无
+        /// </summary>
+        None = 0,
+
+        /// <summary>
+        /// DL/T 645—1997
+        /// </summary>
+        DLT_645_1997 = 1,
+
+        /// <summary>
+        /// 交流采样装置通信协议(电表)
+        /// </summary>
+        ACSamplingDevice = 2,
+
+        /// <summary>
+        /// DL/T 645—2007
+        /// </summary>
+        DLT_645_2007 = 30,
+
+        /// <summary>
+        /// 载波通信
+        /// </summary>
+        Carrierwave = 31,
+
+        /// <summary>
+        /// CJ/T 188—2018协议(水表)
+        /// </summary>
+        CJT_188_2018 = 32,
+
+        /// <summary>
+        /// CJ/T 188—2004协议
+        /// </summary>
+        CJT_188_2004 = 33,
+
+        /// <summary>
+        /// MODBUS-RTU
+        /// </summary>
+        MODBUS_RTU = 34,
+    }
+}

src/JiShe.CollectBus.Domain/Ammeters/AmmeterInfo.cs

@@ -87,7 +87,7 @@ namespace JiShe.CollectBus.Ammeters
         public int TimeDensity { get; set; }
 
         /// <summary>
-        /// 该电表方案下采集项，如：0D_80
+        /// 该电表方案下采集项，JSON格式，如：["0D_80","0D_80"]
         /// </summary>
         public string ItemCodes { get; set; }
 

src/JiShe.CollectBus.Domain/GatherItem/GatherItemInfo.cs

@@ -0,0 +1,21 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.GatherItem
+{
+    public class GatherItemInfo
+    {
+        /// <summary>
+        /// 数据类型
+        /// </summary>
+        public string DataType { get; set; }
+
+        /// <summary>
+        /// 采集项编码
+        /// </summary>
+        public string ItemCode { get; set; }
+    }
+}

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

@@ -224,7 +224,7 @@ namespace JiShe.CollectBus.Host
             {
                 config.SetListenIPHosts(int.Parse(configuration["TCP:ClientPort"] ?? "10500"))
                     //.SetTcpDataHandlingAdapter(()=>new StandardFixedHeaderDataHandlingAdapter())
-                    //.SetGetDefaultNewId(() => Guid.NewGuid().ToString())//定义ClinetId的生成策略
+                    //.SetGetDefaultNewId(() => Guid.NewGuid().ToString())//定义ClientId的生成策略
                     .ConfigurePlugins(a =>
                     {
                         a.Add<TcpCloseMonitor>();

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

@@ -14,6 +14,7 @@ namespace JiShe.CollectBus.Protocol.Contracts
         public const string SubscriberReceivedHeartbeatEventName = "received.heartbeat.event";
         public const string SubscriberReceivedLoginEventName = "received.login.event";
 
+        #region 电表消息主题
         /// <summary>
         /// 1分钟采集电表数据下行消息主题
         /// </summary>
@@ -27,6 +28,18 @@ namespace JiShe.CollectBus.Protocol.Contracts
         /// </summary>
         public const string AmmeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.ammeter.event";
 
+        /// <summary>
+        /// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号、定时阀控等
+        /// </summary>
+        public const string AmmeterSubscriberWorkerOtherIssuedEventName = "issued.other.ammeter.event";
+
+        /// <summary>
+        /// 电表手动阀控
+        /// </summary>
+        public const string AmmeterSubscriberWorkerManualValveControlIssuedEventName = "issued.control.ammeter.event";
+        #endregion
+
+        #region 水表消息主题
         /// <summary>
         /// 1分钟采集水表数据下行消息主题
         /// </summary>
@@ -40,5 +53,18 @@ namespace JiShe.CollectBus.Protocol.Contracts
         /// </summary>
         public const string WatermeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.watermeter.event";
 
+        /// <summary>
+        /// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号、定时阀控等
+        /// </summary>
+        public const string WatermeterSubscriberWorkerOtherIssuedEventName = "issued.other.watermeter.event";
+
+        /// <summary>
+        /// 水表手动阀控
+        /// </summary>
+        public const string WatermeterSubscriberWorkerManualValveControlIssuedEventName = "issued.control.watermeter.event";
+        #endregion
+
+
+
     }
 }