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

JiShe.CollectBus.FreeRedisProvider/FreeRedisProvider/FreeRedisProviderService.cs

@@ -0,0 +1,62 @@
+using FreeRedis;
+using JetBrains.Annotations;
+using Microsoft.Extensions.Options;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Text;
+using System.Text.Json;
+using System.Threading.Tasks;
+using Volo.Abp.DependencyInjection;
+
+namespace JiShe.CollectBus.FreeRedisProvider;
+
+public class FreeRedisProviderService : IFreeRedisProviderService, ISingletonDependency
+{
+    private FreeRedisOptions freeRedisOptions;
+
+    /// <summary>
+    /// FreeRedis 
+    /// </summary>
+    public FreeRedisProviderService(IOptions<FreeRedisOptions> options)
+    {
+        freeRedisOptions = options.Value;
+    }
+
+    [NotNull]
+    public  IRedisClient FreeRedis
+    {
+        get
+        {
+            return GetClient();
+        }
+    }
+
+    /// <summary>
+    /// 获取 FreeRedis 客户端
+    /// </summary>
+    /// <returns></returns>
+    public IRedisClient GetClient()
+    {
+        string connectionString = $"{freeRedisOptions.Configuration},defaultdatabase={freeRedisOptions.DefaultDB}";
+        var redisClinet = new RedisClient(connectionString);
+        redisClinet.Serialize = obj => JsonSerializer.Serialize(obj);
+        redisClinet.Deserialize = (json, type) => JsonSerializer.Deserialize(json, type);
+        redisClinet.Notice += (s, e) => Trace.WriteLine(e.Log);
+
+        return redisClinet;
+    }
+
+    /// <summary>
+    /// 切换Redis数据库
+    /// </summary>
+    /// <param name="index"></param>
+    /// <returns></returns>
+    public IRedisClient GetDatabase(int index = 0)
+    {
+        var redisClinet = GetClient();
+        redisClinet.GetDatabase(index);
+        return redisClinet;
+    }
+}

JiShe.CollectBus.FreeRedisProvider/FreeRedisProvider/IFreeRedisProviderService.cs

@@ -0,0 +1,29 @@
+using FreeRedis;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.FreeRedisProvider;
+
+public interface IFreeRedisProviderService
+{
+    /// <summary>
+    /// 默认客户端
+    /// </summary>
+    IRedisClient FreeRedis { get; }
+
+    /// <summary>
+    /// 获取客户端
+    /// </summary>
+    /// <returns></returns>
+    IRedisClient GetClient();
+
+    /// <summary>
+    /// 切换Redis数据库
+    /// </summary>
+    /// <param name="index"></param>
+    /// <returns></returns>
+    IRedisClient GetDatabase(int index = 0);
+}

JiShe.CollectBus.FreeRedisProvider/FreeRedisProvider/Options/FreeRedisOptions.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.FreeRedisProvider;
+
+public class FreeRedisOptions
+{
+    /// <summary>
+    /// 连接字符串
+    /// </summary>
+    public string? Configuration { get; set; }
+
+    /// <summary>
+    /// 默认数据库
+    /// </summary>
+    public string? DefaultDB { get; set; }
+
+    /// <summary>
+    /// HangfireDB
+    /// </summary>
+    public string? HangfireDB { get; set; }
+}

JiShe.CollectBus.FreeRedisProvider/FreeRedisProviderModule.cs

@@ -0,0 +1,26 @@
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.Modularity;
+
+namespace JiShe.CollectBus.FreeRedisProvider;
+
+public class FreeRedisProviderModule : AbpModule
+{
+    public override void ConfigureServices(ServiceConfigurationContext context)
+    {
+        var configuration = context.Services.GetConfiguration();
+
+        Configure<FreeRedisOptions>(options =>
+        {
+            configuration.GetSection("Redis").Bind(options);
+        });
+
+
+    }
+}
+

JiShe.CollectBus.FreeRedisProvider/JiShe.CollectBus.FreeRedisProvider.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+	<ItemGroup>
+		<PackageReference Include="FreeRedis" Version="1.3.6" />
+		<PackageReference Include="Volo.Abp" Version="8.3.3" />
+	</ItemGroup>
+</Project>

JiShe.CollectBus.sln

@@ -29,6 +29,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JiShe.CollectBus.DbMigrator
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "JiShe.CollectBus.FreeSql", "src\JiShe.CollectBus.FreeSql\JiShe.CollectBus.FreeSql.csproj", "{FE0457D9-4038-4A17-8808-DCAD06CFC0A0}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.FreeRedisProvider", "JiShe.CollectBus.FreeRedisProvider\JiShe.CollectBus.FreeRedisProvider.csproj", "{920445D9-18D2-4886-9053-6A4CC3B4F3E2}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -83,6 +85,10 @@ Global
 		{FE0457D9-4038-4A17-8808-DCAD06CFC0A0}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{FE0457D9-4038-4A17-8808-DCAD06CFC0A0}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{FE0457D9-4038-4A17-8808-DCAD06CFC0A0}.Release|Any CPU.Build.0 = Release|Any CPU
+		{920445D9-18D2-4886-9053-6A4CC3B4F3E2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{920445D9-18D2-4886-9053-6A4CC3B4F3E2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{920445D9-18D2-4886-9053-6A4CC3B4F3E2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{920445D9-18D2-4886-9053-6A4CC3B4F3E2}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -100,6 +106,7 @@ Global
 		{38C1808B-009A-418B-B17B-AB3626341B5D} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 		{8BA01C3D-297D-42DF-BD63-EF07202A0A67} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 		{FE0457D9-4038-4A17-8808-DCAD06CFC0A0} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
+		{920445D9-18D2-4886-9053-6A4CC3B4F3E2} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {4324B3B4-B60B-4E3C-91D8-59576B4E26DD}

src/JiShe.CollectBus.Application.Contracts/CollectBusRemoteServiceConsts.cs

@@ -4,5 +4,5 @@ public class CollectBusRemoteServiceConsts
 {
     public const string RemoteServiceName = "CollectBus";
 
-    public const string ModuleName = "collectBus";
+    public const string ModuleName = "collectBus"; 
 }

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

@@ -0,0 +1,64 @@
+using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.Watermeter;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.Application.Services;
+using Volo.Abp.DependencyInjection;
+
+namespace JiShe.CollectBus.Workers
+{
+    /// <summary>
+    /// 定时任务基础约束
+    /// </summary>
+    public interface IScheduledMeterReadingService : IApplicationService
+    {
+        /// <summary>
+        /// 获取电表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "");
+         
+        /// <summary>
+        /// 初始化电表缓存数据
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        Task InitAmmeterCacheData(string gatherCode = "");
+
+        /// <summary>
+        /// 获取水表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "");
+
+        /// <summary>
+        /// 初始化水表缓存数据
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        Task InitWatermeterCacheData(string gatherCode = "");
+
+        /// <summary>
+        /// 1分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterOneMinuteReading();
+
+        /// <summary>
+        /// 5分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterFiveMinuteReading();
+
+        /// <summary>
+        /// 15分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterFifteenMinuteReading();
+    }
+}

src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs

@@ -0,0 +1,31 @@
+using System.Threading.Tasks;
+using JiShe.CollectBus.Common.Models;
+using JiShe.CollectBus.MessageReceiveds;
+using Volo.Abp.Application.Services;
+
+namespace JiShe.CollectBus.Subscribers
+{
+    /// <summary>
+    /// 定时抄读任务消息订阅
+    /// </summary>
+    public interface IWorkerSubscriberAppService : IApplicationService
+    {
+        /// <summary>
+        /// 1分钟采集电表数据下行消息消费订阅
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage);
+
+        /// <summary>
+        /// 5分钟采集电表数据下行消息消费订阅
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage);
+
+        /// <summary>
+        /// 15分钟采集电表数据下行消息消费订阅
+        /// </summary>
+        /// <returns></returns>
+        Task ScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage);
+    }
+}

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

@@ -1,99 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Ammeters
-{
-    public class AmmeterInfo
-    {
-        public int ID { get; set; }
-        public string Name { get; set; }
-        public int FocusID { get; set; }
-        public string Address { get; set; }
-        public string AreaCode { get; set; }
-        /// <summary>
-        /// 电表类别 （1单相、2三相三线、3三相四线）,
-        /// 07协议： 开合闸指令(1A开闸断电,1C单相表合闸,1B多相表合闸)  645 2007 表
-        /// 97协议：//true(合闸);false(跳闸)  545 1997 没有单相多相  之分  "true" ? "9966" : "3355"
-        /// </summary>		
-        public int TypeName { get; set; }
-        /// <summary>
-        /// 跳合闸状态字段：  0 合闸，1 跳闸
-        /// 电表：TripState （0 合闸-通电, 1 断开、跳闸）；
-        /// </summary>
-        public int TripState { get; set; }
-        /// <summary>
-        /// 规约  -电表default(30) 1：97协议，30：07协议
-        /// </summary>
-        public int? Protocol { get; set; }
-        /// <summary>
-        /// 一个集中器下的[MeteringCode]必须唯一。 PN
-        /// </summary>
-        public int MeteringCode { get; set; }
-        /// <summary>
-        /// 电表通信地址
-        /// </summary>
-        public string AmmerterAddress { get; set; }
-        /// <summary>
-        /// 波特率 default(2400)
-        /// </summary>
-        public int Baudrate { get; set; }
-        /// <summary>
-        ///  MeteringPort 端口就几个可以枚举。
-        /// </summary>
-        public int MeteringPort { get; set; }
-        /// <summary>
-        /// 电表密码
-        /// </summary>		
-        public string Password { get; set; }
-        /// <summary>
-        /// 采集时间间隔(分钟，如15)
-        /// </summary>
-        public int TimeDensity { get; set; }
-        /// <summary>
-        /// 该电表方案下采集项，如：0D_80
-        /// </summary>
-        public string ItemCodes { get; set; }
-        /// <summary>        
-        /// State表状态: 
-        /// 0新装（未下发），1运行(档案下发成功时设置状态值1)， 2暂停, 100销表（销表后是否重新启用）
-        /// 特定：State   -1 已删除
-        /// </summary>		
-        public int State { get; set; }
-        /// <summary>
-        /// 是否自动采集(0:主动采集，1：自动采集)
-        /// </summary>
-        public int AutomaticReport { get; set; }
-        /// <summary>
-        /// 该电表方案下采集项编号
-        /// </summary>
-        public string DataTypes { get; set; }
-        /// <summary>
-        /// 品牌型号
-        /// </summary>
-        public string BrandType { get; set; }
-        /// <summary>
-        /// 采集器编号
-        /// </summary>
-        public string GatherCode { get; set; }
-        /// <summary>
-        /// 是否特殊表
-        /// </summary>
-        public int Special { get; set; }
-        /// <summary>
-        /// 费率类型，单、多   (SingleRate :单费率（单相表1），多费率（其他0） ，与TypeName字段无关)  
-        /// SingleRate ? "单" : "复"  
-        /// [SingleRate]  --0 复费率 false ， 1 单费率 true   （与PayPlanID保持一致）
-        ///对应 TB_PayPlan.Type: 1复费率，2单费率
-        /// </summary>
-        public bool SingleRate { get; set; }
-        public int ProjectID { get; set; }
-        /// <summary>
-        /// 是否异常集中器 0:正常，1异常
-        /// </summary>
-        public int AbnormalState { get; set; }
-        public DateTime LastTime { get; set; }
-    }
-}

src/JiShe.CollectBus.Application/CollectBusAppService.cs

@@ -1,6 +1,11 @@
-using JiShe.CollectBus.FreeSql;
+using FreeRedis;
+using FreeSql;
+using JiShe.CollectBus.FreeRedisProvider;
+using JiShe.CollectBus.FreeSql;
 using JiShe.CollectBus.Localization;
 using Microsoft.AspNetCore.Mvc;
+using System.Collections.Generic;
+using System.Threading.Tasks;
 using Volo.Abp.Application.Services;
 
 namespace JiShe.CollectBus;
@@ -9,10 +14,11 @@ namespace JiShe.CollectBus;
 public abstract class CollectBusAppService : ApplicationService
 {
     public IFreeSqlProvider SqlProvider => LazyServiceProvider.LazyGetRequiredService<IFreeSqlProvider>();
+    protected IFreeRedisProviderService FreeRedisProvider => LazyServiceProvider.LazyGetService<IFreeRedisProviderService>()!;
 
     protected CollectBusAppService()
     {
         LocalizationResource = typeof(CollectBusResource);
         ObjectMapperContext = typeof(CollectBusApplicationModule);
-    }
+    } 
 }

src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs

@@ -10,6 +10,7 @@ using System.Reflection;
 using JiShe.CollectBus.FreeSql;
 using System;
 using Volo.Abp.AspNetCore.Mvc.AntiForgery;
+using JiShe.CollectBus.FreeRedisProvider;
 
 namespace JiShe.CollectBus;
 
@@ -19,6 +20,7 @@ namespace JiShe.CollectBus;
     typeof(AbpDddApplicationModule),
     typeof(AbpAutoMapperModule),
     typeof(AbpBackgroundWorkersModule),
+    typeof(FreeRedisProviderModule),
     typeof(CollectBusFreeSqlModule)
     )]
 public class CollectBusApplicationModule : AbpModule

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

@@ -23,6 +23,7 @@
 		<PackageReference Include="TouchSocket" Version="2.1.9" />
     <PackageReference Include="TouchSocket.Hosting" Version="2.1.9" />
     <PackageReference Include="DotNetCore.CAP" Version="8.3.1" />
+    <ProjectReference Include="..\..\JiShe.CollectBus.FreeRedisProvider\JiShe.CollectBus.FreeRedisProvider.csproj" />
 
 		<ProjectReference Include="..\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.csproj" />
     <ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />

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

@@ -53,14 +53,17 @@ namespace JiShe.CollectBus.Plugins
             var aFn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.AFN);
             var fn = (int?)hexStringList.GetAnalyzeValue(CommandChunkEnum.FN);
             var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A);
-            if (aFn.HasValue && fn.HasValue)
+            if (aFn.HasValue && fn.HasValue && aTuple != null && !string.IsNullOrWhiteSpace(aTuple.Item1))
             {
+                string oldClinetId = client.Id;
+                await client.ResetIdAsync(aTuple.Item1);
+
                 if ((AFN)aFn == AFN.链路接口检测)
                 {
                     switch (fn)
                     {
                         case 1:
-                            await OnTcpLoginReceived(client, messageHexString, aTuple.Item1);
+                            await OnTcpLoginReceived(client, messageHexString, aTuple.Item1, oldClinetId);
                             break;
                         case 3:
                             await OnTcpHeartbeatReceived(client, messageHexString, aTuple.Item1);
@@ -113,7 +116,16 @@ namespace JiShe.CollectBus.Plugins
 
             await e.InvokeNext();
         }
-        private async Task OnTcpLoginReceived(ITcpSessionClient client, string messageHexString, string deviceNo)
+
+        /// <summary>
+        /// 登录帧处理
+        /// </summary>
+        /// <param name="client"></param>
+        /// <param name="messageHexString"></param>
+        /// <param name="deviceNo">集中器编号</param>
+        /// <param name="oldClinetId">TCP首次连接时的Id</param>
+        /// <returns></returns>
+        private async Task OnTcpLoginReceived(ITcpSessionClient client, string messageHexString, string deviceNo,string oldClinetId)
         {
             var messageReceivedLoginEvent = new MessageReceivedLogin
             {

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

@@ -0,0 +1,137 @@
+using FreeRedis;
+using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.Common.Consts;
+using JiShe.CollectBus.Watermeter;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.Application.Services;
+
+namespace JiShe.CollectBus.Workers
+{
+    /// <summary>
+    /// 定时采集服务
+    /// </summary>
+    public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
+    {
+        /// <summary>
+        /// 系统类型
+        /// </summary>
+        public abstract string SystemType { get; }
+
+        /// <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 meterInfoGroup = meterInfos.GroupBy(x => x.FocusAddress).ToList();
+            foreach (var item in meterInfoGroup)
+            {
+                if (string.IsNullOrWhiteSpace(item.Key))
+                {
+                    continue;
+                }
+
+                var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy)}{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);
+            }
+        }
+
+        /// <summary>
+        /// 获取水表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        public virtual async 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 meterInfoGroup = meterInfos.GroupBy(x => x.FocusAddress).ToList();
+            foreach (var item in meterInfoGroup)
+            {
+                if (string.IsNullOrWhiteSpace(item.Key))
+                {
+                    continue;
+                }
+
+                var redisCacheKey = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy)}{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);
+            }
+        }
+
+        /// <summary>
+        /// 1分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual Task ScheduledMeterOneMinuteReading()
+        {
+            throw new NotImplementedException($"{nameof(ScheduledMeterOneMinuteReading)}请根据不同系统类型进行实现");
+        }
+
+        /// <summary>
+        /// 5分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual Task ScheduledMeterFiveMinuteReading()
+        {
+            throw new NotImplementedException($"{nameof(ScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现");
+        }
+
+        /// <summary>
+        /// 15分钟采集电表数据
+        /// </summary>
+        /// <returns></returns>
+        public virtual Task ScheduledMeterFifteenMinuteReading()
+        {
+            throw new NotImplementedException($"{nameof(ScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现");
+        }
+    }
+}

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

@@ -0,0 +1,103 @@
+using FreeRedis;
+using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.Common.Consts;
+using JiShe.CollectBus.FreeRedisProvider;
+using JiShe.CollectBus.FreeSql;
+using JiShe.CollectBus.Watermeter;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.DependencyInjection;
+
+namespace JiShe.CollectBus.Workers
+{
+    /// <summary>
+    /// 能耗系统定时采集服务
+    /// </summary>
+    [AllowAnonymous]
+    //[Route($"/energy/app/scheduled")]
+    public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
+    {
+        public sealed override string SystemType => SystemTypeConst.Energy;
+
+        /// <summary>
+        /// 获取电表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        //[HttpGet]
+        //[Route($"ammeter/list")]
+        public override async Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "V4-Gather-8890")
+        {
+            string sql = $@"SELECT C.ID,C.Name,C.FocusID,C.SingleRate,C.MeteringCode,C.Code AS BrandType,C.Baudrate,C.Password,C.MeteringPort,C.[Address] AS AmmerterAddress,C.TypeName,C.Protocol,C.TripState,C.[State],B.[Address],B.AreaCode,B.AutomaticReport,D.DataTypes,B.TimeDensity,A.GatherCode,C.Special,C.[ProjectID],B.AbnormalState,B.LastTime,CONCAT(B.AreaCode, B.[Address]) AS FocusAddress
+              FROM TB_GatherInfo(NOLOCK) AS A
+              INNER JOIN TB_FocusInfo(NOLOCK) AS B ON A.ID = B.GatherInfoID AND B.RemoveState >= 0 AND B.State>=0
+              INNER JOIN TB_AmmeterInfo(NOLOCK) AS C ON B.ID = C.FocusID AND C.State>= 0 AND C.State<100
+              INNER JOIN TB_AmmeterGatherItem(NOLOCK) AS D ON C.ID = D.AmmeterID AND D.State>=0
+              WHERE 1=1 ";
+
+            if (!string.IsNullOrWhiteSpace(gatherCode))
+            {
+                sql = $@"{sql} A.GatherCode = '{gatherCode}'";
+            }
+            return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
+                .Ado
+                .QueryAsync<AmmeterInfo>(sql);
+        }
+
+        /// <summary>
+        /// 获取水表信息
+        /// </summary>
+        /// <param name="gatherCode">采集端Code</param>
+        /// <returns></returns>
+        //[HttpGet]
+        //[Route($"ammeter/list")]
+        public override async Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "V4-Gather-8890")
+        {
+            string sql = $@"SELECT 
+                            A.ID,
+                            A.Name,
+                            A.FocusID,
+                            A.MeteringCode,
+                            A.Baudrate,
+                            A.MeteringPort,
+                            A.[Address] AS MeterAddress,
+                            A.[Password],
+                            A.TypeName,
+                            A.Protocol,
+                            A.Code,
+                            A.LinkType,
+                            A.HaveValve,
+                            A.MeterType AS MeterTypeName, 
+                            A.MeterBrand,
+                            A.TimesRate,
+                            A.TimeDensity,
+                            A.TripState,
+                            B.[Address],
+                            B.AreaCode,
+                            B.AutomaticReport,
+                            A.[State],
+                            C.GatherCode,
+                            A.[ProjectID],
+                            B.AbnormalState,
+                            B.LastTime
+                            FROM [dbo].[TB_WatermeterInfo](NOLOCK) AS A 
+                            INNER JOIN [dbo].[TB_FocusInfo](NOLOCK) AS B ON A.FocusID=B.ID AND B.RemoveState >= 0 AND B.State>=0
+                            INNER JOIN [dbo].[TB_GatherInfo](NOLOCK) AS C ON B.GatherInfoID=C.ID
+                            WHERE A.State>=0 AND A.State<100 ";
+
+            if (!string.IsNullOrWhiteSpace(gatherCode))
+            {
+                sql = $@"{sql} AND C.GatherCode= '{gatherCode}'";
+            }
+            return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
+            .Ado
+            .QueryAsync<WatermeterInfo>(sql);
+        }
+
+    }
+}

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

@@ -22,7 +22,7 @@ namespace JiShe.CollectBus.Workers
 
         public override Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
         {
-            _logger.LogInformation("Executed MyLogWorker..!");
+            _logger.LogError("Executed MyLogWorker..!");
             return Task.CompletedTask;
         }
     }

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

@@ -0,0 +1,111 @@
+using System;
+using System.Threading.Tasks;
+using DeviceDetectorNET.Parser.Device;
+using DotNetCore.CAP;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.Common.Models;
+using JiShe.CollectBus.Devices;
+using JiShe.CollectBus.MessageReceiveds;
+using JiShe.CollectBus.Protocol.Contracts;
+using JiShe.CollectBus.Protocol.Contracts.Interfaces;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using TouchSocket.Sockets;
+using Volo.Abp.Caching;
+using Volo.Abp.Domain.Repositories;
+
+namespace JiShe.CollectBus.Subscribers
+{
+    /// <summary>
+    /// 定时抄读任务消息消费订阅
+    /// </summary>
+    [Route($"/worker/app/subscriber")]
+    public class WorkerSubscriberAppService : CollectBusAppService, IWorkerSubscriberAppService,ICapSubscribe
+    {
+        private readonly ILogger<WorkerSubscriberAppService> _logger;
+        private readonly ITcpService _tcpService;
+        private readonly IServiceProvider _serviceProvider;
+
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="WorkerSubscriberAppService"/> class.
+        /// </summary>
+        /// <param name="logger">The logger.</param>
+        /// <param name="tcpService">The TCP service.</param>
+        /// <param name="serviceProvider">The service provider.</param>
+        public WorkerSubscriberAppService(ILogger<WorkerSubscriberAppService> logger, 
+            ITcpService tcpService, IServiceProvider serviceProvider)
+        {
+            _logger = logger;
+            _tcpService = tcpService;
+            _serviceProvider = serviceProvider; 
+        }
+
+        /// <summary>
+        /// 一分钟定时抄读任务消息消费订阅
+        /// </summary>
+        /// <param name="receivedMessage"></param>
+        /// <returns></returns>
+        [HttpPost]
+        [Route("oneminute/issued-event")]
+        [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)]
+        public async Task ScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage)
+        {
+            _logger.LogInformation("1分钟采集电表数据下行消息消费队列开始处理");
+            var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
+            if (protocolPlugin == null)
+            {
+                _logger.LogError("【1分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
+            }
+            else
+            {
+                await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message);
+            }
+        }
+
+        /// <summary>
+        /// 5分钟采集电表数据下行消息消费订阅
+        /// </summary>
+        /// <param name="receivedMessage"></param>
+        /// <returns></returns>
+        [HttpPost]
+        [Route("fiveminute/issued-event")]
+        [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)]
+        public async Task ScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage)
+        {
+            _logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理");
+            var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
+            if (protocolPlugin == null)
+            {
+                _logger.LogError("【5分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
+            }
+            else
+            {
+                await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message);
+            }
+        }
+
+        /// <summary>
+        /// 15分钟采集电表数据下行消息消费订阅
+        /// </summary>
+        /// <param name="receivedMessage"></param>
+        /// <returns></returns>
+        [HttpPost]
+        [Route("fifteenminute/issued-event")]
+        [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)]
+        public async Task ScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage)
+        {
+            _logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");
+            var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
+            if (protocolPlugin == null)
+            {
+                _logger.LogError("【15分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
+            }
+            else
+            {
+                await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Consts
+{
+    public class FreeRedisConst
+    {
+        /// <summary>
+        /// 缓存基础目录
+        /// </summary>
+        public const string CacheBasicDirectoryKey = "CollectBus:";
+
+        /// <summary>
+        /// 缓存电表信息
+        /// </summary>
+        public const string CacheAmmeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:AmmeterInfo:";
+
+        /// <summary>
+        /// 缓存水表信息
+        /// </summary>
+        public const string CacheWatermeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:WatermeterInfo:";
+    }
+}

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

@@ -0,0 +1,24 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Consts
+{
+    /// <summary>
+    /// 系统类型常量
+    /// </summary>
+    public class SystemTypeConst
+    {
+        /// <summary>
+        /// 预付费系统
+        /// </summary>
+        public const string Prepay = "Prepay";
+
+        /// <summary>
+        /// 能耗系统
+        /// </summary>
+        public const string Energy = "Energy";
+    }
+}

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

@@ -0,0 +1,53 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Enums
+{
+    /// <summary>
+    /// 表计类型
+    /// 电表= 1,水表= 2,燃气表= 3,热能表= 4,水表流量计=5，燃气表流量计=6
+    /// </summary>
+    public enum MeterTypeEnum
+    {
+        /// <summary>
+        /// 电表
+        /// </summary>
+        Ammeter = 1,
+        /// <summary>
+        /// 水表
+        /// </summary>
+        WaterMeter = 2,
+        /// <summary>
+        /// 燃气表
+        /// </summary>
+        Gasmeter = 3,
+        /// <summary>
+        /// 热能表
+        /// </summary>
+        HeatMeter = 4,
+        /// <summary>
+        /// 水表流量计
+        /// </summary>
+        WaterMeterFlowmeter = 5,
+        /// <summary>
+        /// 燃气表流量计
+        /// </summary>
+        GasmeterFlowmeter = 6,
+        /// <summary>
+        /// 特殊电表
+        /// </summary>
+        SpecialAmmeter = 7,
+        /// <summary>
+        /// 传感器
+        /// </summary>
+        Sensor = 8,
+
+        /// <summary>
+        /// 采集器
+        /// </summary>
+        Collector = 9,
+    }
+}

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

@@ -8,80 +8,121 @@ namespace JiShe.CollectBus.Ammeters
 {
     public class AmmeterInfo
     {
+        /// <summary>
+        /// 电表ID
+        /// </summary>
         public int ID { get; set; }
+
+        /// <summary>
+        /// 电表名称
+        /// </summary>
         public string Name { get; set; }
+
+        /// <summary>
+        /// 集中器ID
+        /// </summary>
         public int FocusID { get; set; }
+
+        /// <summary>
+        /// 集中器地址
+        /// </summary>
+        public string FocusAddress { get; set; }
+
+        /// <summary>
+        /// 集中器地址
+        /// </summary>
         public string Address { get; set; }
+
+        /// <summary>
+        /// 集中器区域代码
+        /// </summary>
         public string AreaCode { get; set; }
+
         /// <summary>
         /// 电表类别 （1单相、2三相三线、3三相四线）,
         /// 07协议： 开合闸指令(1A开闸断电,1C单相表合闸,1B多相表合闸)  645 2007 表
         /// 97协议：//true(合闸);false(跳闸)  545 1997 没有单相多相  之分  "true" ? "9966" : "3355"
         /// </summary>		
         public int TypeName { get; set; }
+
         /// <summary>
         /// 跳合闸状态字段：  0 合闸，1 跳闸
         /// 电表：TripState （0 合闸-通电, 1 断开、跳闸）；
         /// </summary>
         public int TripState { get; set; }
+
         /// <summary>
         /// 规约  -电表default(30) 1：97协议，30：07协议
         /// </summary>
         public int? Protocol { get; set; }
+
         /// <summary>
         /// 一个集中器下的[MeteringCode]必须唯一。 PN
         /// </summary>
         public int MeteringCode { get; set; }
+
         /// <summary>
         /// 电表通信地址
         /// </summary>
         public string AmmerterAddress { get; set; }
+
         /// <summary>
         /// 波特率 default(2400)
         /// </summary>
         public int Baudrate { get; set; }
+
         /// <summary>
         ///  MeteringPort 端口就几个可以枚举。
         /// </summary>
         public int MeteringPort { get; set; }
+
         /// <summary>
         /// 电表密码
         /// </summary>		
         public string Password { get; set; }
+
         /// <summary>
         /// 采集时间间隔(分钟，如15)
         /// </summary>
         public int TimeDensity { get; set; }
+
         /// <summary>
         /// 该电表方案下采集项，如：0D_80
         /// </summary>
         public string ItemCodes { get; set; }
+
         /// <summary>        
         /// State表状态: 
         /// 0新装（未下发），1运行(档案下发成功时设置状态值1)， 2暂停, 100销表（销表后是否重新启用）
         /// 特定：State   -1 已删除
         /// </summary>		
         public int State { get; set; }
+
         /// <summary>
         /// 是否自动采集(0:主动采集，1：自动采集)
         /// </summary>
         public int AutomaticReport { get; set; }
+
         /// <summary>
         /// 该电表方案下采集项编号
         /// </summary>
         public string DataTypes { get; set; }
+
         /// <summary>
         /// 品牌型号
         /// </summary>
         public string BrandType { get; set; }
+
         /// <summary>
         /// 采集器编号
         /// </summary>
         public string GatherCode { get; set; }
+
         /// <summary>
         /// 是否特殊表
         /// </summary>
         public int Special { get; set; }
+
         /// <summary>
         /// 费率类型，单、多   (SingleRate :单费率（单相表1），多费率（其他0） ，与TypeName字段无关)  
         /// SingleRate ? "单" : "复"  
@@ -89,11 +130,17 @@ namespace JiShe.CollectBus.Ammeters
         ///对应 TB_PayPlan.Type: 1复费率，2单费率
         /// </summary>
         public bool SingleRate { get; set; }
+
+        /// <summary>
+        /// 项目ID
+        /// </summary>
         public int ProjectID { get; set; }
+
         /// <summary>
         /// 是否异常集中器 0:正常，1异常
         /// </summary>
         public int AbnormalState { get; set; }
-        public DateTime LastTime { get; set; }
+
+        public DateTime LastTime { get; set; } 
     }
 }

src/JiShe.CollectBus.Domain/Devices/Device.cs

@@ -27,16 +27,34 @@ namespace JiShe.CollectBus.Devices
             Status = status;
         }
 
+        /// <summary>
+        /// 集中器编号，在集中器登录时解析获取，并会更新为当前TCP连接的最新ClientId
+        /// </summary>
         public string Number { get; set; }
 
+        /// <summary>
+        /// 首次上线时间
+        /// </summary>
         public DateTime FirstOnlineTime { get; set; }
 
+        /// <summary>
+        /// 最后上线时间
+        /// </summary>
         public DateTime LastOnlineTime { get; set; }
 
+        /// <summary>
+        /// TCP客户端首次连接ID，在登录解析成功以后会被Number集中器编号覆盖
+        /// </summary>
         public string ClientId { get; set; }
 
+        /// <summary>
+        /// TCP客户端断线时间，用于计算是否断线
+        /// </summary>
         public DateTime? LastOfflineTime { get; set; }
 
+        /// <summary>
+        /// 设备状态
+        /// </summary>
         public DeviceStatus Status { get; set; }
 
         public void UpdateByLoginAndHeartbeat(string clientId)

src/JiShe.CollectBus.Domain/Watermeter/WatermeterInfo.cs

@@ -0,0 +1,158 @@
+using JiShe.CollectBus.Common.Enums;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Watermeter
+{
+    /// <summary>
+    /// 水表信息
+    /// </summary>
+    public class WatermeterInfo
+    {
+        /// <summary>
+        /// 水表ID
+        /// </summary>
+        public int ID { get; set; }
+
+        /// <summary>
+        /// 水表名称
+        /// </summary>
+        public string Name { get; set; }
+        /// <summary>
+        /// 表密码
+        /// </summary>		
+        public string Password { get; set; }
+
+        /// <summary>
+        /// 集中器ID
+        /// </summary>
+        public int FocusID { get; set; }
+
+        /// <summary>
+        /// 集中器地址
+        /// </summary>
+        public string FocusAddress { get; set; }
+
+        /// <summary>
+        /// 一个集中器下的[MeteringCode]必须唯一。 PN
+        /// </summary>
+        public int MeteringCode { get; set; }
+
+        /// <summary>
+        /// 波特率 default(2400)
+        /// </summary>
+        public int Baudrate { get; set; }
+
+        /// <summary>
+        ///  MeteringPort 端口就几个可以枚举。
+        /// </summary>
+        public int MeteringPort { get; set; }
+
+        /// <summary>
+        /// 水表通信地址 (当protocol=32时，Address为14位字符；否则12位字符)
+        /// </summary>		
+        public string MeterAddress { get; set; }
+
+        /// <summary>
+        /// 水表类别 （水表类型改成“公称口径”）
+        /// </summary>
+        public string TypeName { get; set; }
+
+        /// <summary>
+        /// 规约  -电表default(30) 1：97协议，30：07协议,32:CJ/T 188—2018协议
+        /// </summary>
+        public int? Protocol { get; set; }
+
+
+        public string Code { get; set; }
+        /// <summary>
+        /// 通讯方案：
+        /// NB-IOT常德水表、NB-IOT泽联电表、GPRS华立水表、
+        /// RS-485、无线、载波 
+        /// </summary>
+        public string LinkType { get; set; }
+
+        /// <summary>
+        /// HaveValve:  是否带阀   （0 不带阀， 1 带阀）
+        /// 注意：NULL表示未设置
+        /// </summary>
+        public int? HaveValve { get; set; }
+
+        /// <summary>
+        /// 设备类型: 水表\气表、流量计
+        /// </summary>
+        public string MeterTypeName { get; set; }
+
+        /// <summary>
+        /// 表计类型
+        //// 电表= 1,水表= 2,燃气表= 3,热能表= 4,水表流量计=5，燃气表流量计=6,特殊电表=7
+        /// </summary>
+        public MeterTypeEnum MeterType { get; set; }
+        /// <summary>
+        /// 设备品牌；
+        /// (当 MeterType = 水表， 如 威铭、捷先 等)
+        /// (当 MeterType = 流量计, 如 西恩超声波流量计、西恩电磁流量计、涡街流量计 等)
+        /// </summary>
+        public string MeterBrand { get; set; }
+
+        /// <summary>
+        /// 倍率
+        /// </summary>
+        public decimal TimesRate { get; set; }
+
+        /// <summary>
+        /// 集中器地址
+        /// </summary>
+        public string Address { get; set; }
+
+        /// <summary>
+        /// 网关地址
+        /// </summary>
+        public string GateAddress { get; set; }
+        /// <summary>
+        /// 集中器区域
+        /// </summary>
+        public string AreaCode { get; set; }
+        /// <summary>
+        /// 通讯状态        
+        /// 水表：TripState （0 合闸-开阀, 1 关阀）；开阀关阀
+        /// </summary>
+        public int TripState { get; set; }
+        /// <summary>
+        /// 是否自动采集
+        /// </summary>
+        public int AutomaticReport { get; set; }
+        /// <summary>        
+        /// State表状态: 
+        /// 0新装（未下发），1运行(档案下发成功时设置状态值1)， 2暂停, 100销表（销表后是否重新启用）
+        /// 特定：State   -1 已删除
+        /// </summary>		
+        public int State { get; set; }
+        /// <summary>
+        /// 采集时间间隔(分钟，如15)
+        /// </summary>
+        public int TimeDensity { get; set; }
+        /// <summary>
+        /// 采集器编号
+        /// </summary>
+        public string GatherCode { get; set; }
+       
+        /// <summary>
+        /// 项目ID
+        /// </summary>
+        public int ProjectID { get; set; }
+
+        /// <summary>
+        /// 是否异常集中器 0:正常，1异常
+        /// </summary>
+        public int AbnormalState { get; set; }
+
+        /// <summary>
+        /// 集中器最后在线时间
+        /// </summary>
+        public DateTime LastTime { get; set; }
+    }
+}

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

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

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

@@ -14,5 +14,18 @@ namespace JiShe.CollectBus.Protocol.Contracts
         public const string SubscriberReceivedHeartbeatEventName = "received.heartbeat.event";
         public const string SubscriberReceivedLoginEventName = "received.login.event";
 
+        /// <summary>
+        /// 1分钟采集电表数据下行消息主题
+        /// </summary>
+        public const string SubscriberWorkerOneMinuteIssuedEventName = "issued.oneminute.event";
+        /// <summary>
+        /// 5分钟采集电表数据下行消息主题
+        /// </summary>
+        public const string SubscriberWorkerFiveMinuteIssuedEventName = "issued.fiveminute.event";
+        /// <summary>
+        /// 15分钟采集电表数据下行消息主题
+        /// </summary>
+        public const string SubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteenminute.event";
+
     }
 }