diff --git a/src/JiShe.CollectBus.Application.Contracts/Workers/IScheduledMeterReadingService.cs b/src/JiShe.CollectBus.Application.Contracts/Workers/IScheduledMeterReadingService.cs index 7591f3e..bd1a090 100644 --- a/src/JiShe.CollectBus.Application.Contracts/Workers/IScheduledMeterReadingService.cs +++ b/src/JiShe.CollectBus.Application.Contracts/Workers/IScheduledMeterReadingService.cs @@ -15,13 +15,15 @@ namespace JiShe.CollectBus.Workers /// public interface IScheduledMeterReadingService : IApplicationService { + + #region 电表采集处理 /// /// 获取电表信息 /// /// 采集端Code /// Task> GetAmmeterInfoList(string gatherCode = ""); - + /// /// 初始化电表缓存数据 /// @@ -29,6 +31,28 @@ namespace JiShe.CollectBus.Workers /// Task InitAmmeterCacheData(string gatherCode = ""); + /// + /// 1分钟采集电表数据 + /// + /// + Task AmmeterScheduledMeterOneMinuteReading(); + + /// + /// 5分钟采集电表数据 + /// + /// + Task AmmeterScheduledMeterFiveMinuteReading(); + + /// + /// 15分钟采集电表数据 + /// + /// + Task AmmeterScheduledMeterFifteenMinuteReading(); + + #endregion + + + #region 水表采集处理 /// /// 获取水表信息 /// @@ -44,21 +68,24 @@ namespace JiShe.CollectBus.Workers Task InitWatermeterCacheData(string gatherCode = ""); /// - /// 1分钟采集电表数据 + /// 1分钟采集水表数据 /// /// - Task ScheduledMeterOneMinuteReading(); + Task WatermeterScheduledMeterOneMinuteReading(); /// - /// 5分钟采集电表数据 + /// 5分钟采集水表数据 /// /// - Task ScheduledMeterFiveMinuteReading(); + Task WatermeterScheduledMeterFiveMinuteReading(); /// - /// 15分钟采集电表数据 + /// 15分钟采集水表数据 /// /// - Task ScheduledMeterFifteenMinuteReading(); + Task WatermeterScheduledMeterFifteenMinuteReading(); + #endregion + + } } diff --git a/src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs b/src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs index c7442de..dede8c4 100644 --- a/src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs +++ b/src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs @@ -10,22 +10,45 @@ namespace JiShe.CollectBus.Subscribers /// public interface IWorkerSubscriberAppService : IApplicationService { + + #region 电表消息采集 /// /// 1分钟采集电表数据下行消息消费订阅 /// /// - Task ScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + Task AmmeterScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); /// /// 5分钟采集电表数据下行消息消费订阅 /// /// - Task ScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + Task AmmeterScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); /// /// 15分钟采集电表数据下行消息消费订阅 /// /// - Task ScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + Task AmmeterScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + #endregion + + #region 水表消息采集 + /// + /// 1分钟采集水表数据下行消息消费订阅 + /// + /// + Task WatermeterScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + + /// + /// 5分钟采集水表数据下行消息消费订阅 + /// + /// + Task WatermeterScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + + /// + /// 15分钟采集水表数据下行消息消费订阅 + /// + /// + Task WatermeterScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage issuedEventMessage); + #endregion } } diff --git a/src/JiShe.CollectBus.Application/Workers/BasicScheduledMeterReadingService.cs b/src/JiShe.CollectBus.Application/Workers/BasicScheduledMeterReadingService.cs index 51b6652..e5494a1 100644 --- a/src/JiShe.CollectBus.Application/Workers/BasicScheduledMeterReadingService.cs +++ b/src/JiShe.CollectBus.Application/Workers/BasicScheduledMeterReadingService.cs @@ -2,6 +2,7 @@ using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Watermeter; +using Microsoft.Extensions.Logging; using System; using System.Collections.Generic; using System.Linq; @@ -17,11 +18,19 @@ namespace JiShe.CollectBus.Workers /// public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService { + private readonly ILogger _logger; + public BasicScheduledMeterReadingService(ILogger logger) + { + _logger = logger; + } + /// /// 系统类型 /// public abstract string SystemType { get; } + #region 电表采集处理 + /// /// 获取电表信息 /// @@ -45,32 +54,115 @@ namespace JiShe.CollectBus.Workers throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化电表缓存数据时，电表数据为空"); } - //将表计信息根据集中器分组 - var meterInfoGroup = meterInfos.GroupBy(x => x.FocusAddress).ToList(); - foreach (var item in meterInfoGroup) + //根据采集频率分组 + var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity); + foreach (var itemTimeDensity in meterInfoGroupByTimeDensity) { - if (string.IsNullOrWhiteSpace(item.Key)) + //将表计信息根据集中器分组 + var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList(); + foreach (var item in meterInfoGroup) { - continue; - } + if (string.IsNullOrWhiteSpace(item.Key)) + { + continue; + } - var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy)}{item.Key}"; - Dictionary keyValuePairs = new Dictionary(); - foreach (var subItem in item) - { + var redisCacheKey = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}"; + Dictionary keyValuePairs = new Dictionary(); + foreach (var subItem in item) + { - keyValuePairs.TryAdd($"{subItem.ID}", subItem); + keyValuePairs.TryAdd($"{subItem.ID}", subItem); + } + await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs); } - await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs); } + + _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成"); } + /// + /// 1分钟采集电表数据 + /// + /// + 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(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102"); + return; + } + + //通过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 = FreeRedisProvider.FreeRedis.Eval(luaScript, oneMinutekeyList); // 传递 KEYS + if (oneMinuteAmmerterResult == null) + { + _logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102"); + return; + } + + // 解析结果（结果为嵌套数组） + var parsedResults = new Dictionary>(); + if (oneMinuteAmmerterResult is object[] arr) + { + foreach (object[] item in arr) + { + string key = (string)item[0]; + object[] fieldsAndValues = (object[])item[1]; + + var dict = new Dictionary(); + for (int i = 0; i < fieldsAndValues.Length; i += 2) + { + string field = (string)fieldsAndValues[i]; + string value = (string)fieldsAndValues[i + 1]; + dict[field] = value; + } + parsedResults[key] = dict; + } + } + + _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理完成"); + + } + + /// + /// 5分钟采集电表数据 + /// + /// + public virtual Task AmmeterScheduledMeterFiveMinuteReading() + { + throw new NotImplementedException($"{nameof(AmmeterScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现"); + } + + /// + /// 15分钟采集电表数据 + /// + /// + public virtual Task AmmeterScheduledMeterFifteenMinuteReading() + { + throw new NotImplementedException($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现"); + } + #endregion + + + #region 水表采集处理 + /// /// 获取水表信息 /// /// 采集端Code /// - public virtual async Task> GetWatermeterInfoList(string gatherCode = "") + public virtual Task> GetWatermeterInfoList(string gatherCode = "") { throw new NotImplementedException($"{nameof(GetWatermeterInfoList)}请根据不同系统类型进行实现"); } @@ -85,55 +177,66 @@ namespace JiShe.CollectBus.Workers var meterInfos = await GetWatermeterInfoList(gatherCode); if (meterInfos == null || meterInfos.Count <= 0) { - throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化水表缓存数据时，水表数据为空"); + throw new NullReferenceException($"{nameof(InitWatermeterCacheData)} 初始化水表缓存数据时，水表数据为空"); } - - //将表计信息根据集中器分组 - var meterInfoGroup = meterInfos.GroupBy(x => x.FocusAddress).ToList(); - foreach (var item in meterInfoGroup) + //根据采集频率分组 + var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity); + foreach (var itemTimeDensity in meterInfoGroupByTimeDensity) { - if (string.IsNullOrWhiteSpace(item.Key)) + //将表计信息根据集中器分组 + var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList(); + foreach (var item in meterInfoGroup) { - continue; - } + if (string.IsNullOrWhiteSpace(item.Key)) + { + continue; + } - var redisCacheKey = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy)}{item.Key}"; - Dictionary keyValuePairs = new Dictionary(); - foreach (var subItem in item) - { + var redisCacheKey = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, itemTimeDensity.Key)}{item.Key}"; + Dictionary keyValuePairs = new Dictionary(); + foreach (var subItem in item) + { - keyValuePairs.TryAdd($"{subItem.ID}", subItem); + keyValuePairs.TryAdd($"{subItem.ID}", subItem); + } + await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs); } - await FreeRedisProvider.FreeRedis.HSetAsync(redisCacheKey, keyValuePairs); } + _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化水表缓存数据完成"); } /// - /// 1分钟采集电表数据 + /// 1分钟采集水表数据 /// /// - public virtual Task ScheduledMeterOneMinuteReading() + public virtual async Task WatermeterScheduledMeterOneMinuteReading() { - - throw new NotImplementedException($"{nameof(ScheduledMeterOneMinuteReading)}请根据不同系统类型进行实现"); + //获取缓存中的电表信息 + var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*"; + var oneMinuteList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList); + + _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成"); } /// /// 5分钟采集电表数据 /// /// - public virtual Task ScheduledMeterFiveMinuteReading() + public virtual Task WatermeterScheduledMeterFiveMinuteReading() { - throw new NotImplementedException($"{nameof(ScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现"); + + throw new NotImplementedException($"{nameof(WatermeterScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现"); } /// /// 15分钟采集电表数据 /// /// - public virtual Task ScheduledMeterFifteenMinuteReading() + public virtual Task WatermeterScheduledMeterFifteenMinuteReading() { - throw new NotImplementedException($"{nameof(ScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现"); + throw new NotImplementedException($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现"); } + #endregion + } } diff --git a/src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs b/src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs index 1db9e4a..a5d5940 100644 --- a/src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs +++ b/src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs @@ -7,6 +7,7 @@ using JiShe.CollectBus.FreeSql; using JiShe.CollectBus.Watermeter; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.Logging; using System; using System.Collections.Generic; using System.Linq; @@ -25,9 +26,12 @@ namespace JiShe.CollectBus.Workers public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService { private readonly IRepository _deviceRepository; - public EnergySystemScheduledMeterReadingService(IRepository deviceRepository) + private readonly ILogger _logger; + + public EnergySystemScheduledMeterReadingService(IRepository deviceRepository, ILogger logger):base(logger) { this._deviceRepository = deviceRepository; + this._logger = logger; } public sealed override string SystemType => SystemTypeConst.Energy; diff --git a/src/JiShe.CollectBus.Application/Workers/WorkerSubscriberAppService.cs b/src/JiShe.CollectBus.Application/Workers/WorkerSubscriberAppService.cs index 6c7bfab..552f964 100644 --- a/src/JiShe.CollectBus.Application/Workers/WorkerSubscriberAppService.cs +++ b/src/JiShe.CollectBus.Application/Workers/WorkerSubscriberAppService.cs @@ -26,6 +26,7 @@ namespace JiShe.CollectBus.Subscribers private readonly ILogger _logger; private readonly ITcpService _tcpService; private readonly IServiceProvider _serviceProvider; + private readonly IRepository _deviceRepository; /// @@ -35,22 +36,27 @@ namespace JiShe.CollectBus.Subscribers /// The TCP service. /// The service provider. public WorkerSubscriberAppService(ILogger logger, - ITcpService tcpService, IServiceProvider serviceProvider) + ITcpService tcpService, + IRepository deviceRepository, + IServiceProvider serviceProvider) { _logger = logger; _tcpService = tcpService; - _serviceProvider = serviceProvider; + _serviceProvider = serviceProvider; + _deviceRepository = deviceRepository; } + + #region 电表消息采集 /// /// 一分钟定时抄读任务消息消费订阅 /// /// /// [HttpPost] - [Route("oneminute/issued-event")] - [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)] - public async Task ScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + [Route("ammeter/oneminute/issued-event")] + [CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task AmmeterScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) { _logger.LogInformation("1分钟采集电表数据下行消息消费队列开始处理"); var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); @@ -60,7 +66,12 @@ namespace JiShe.CollectBus.Subscribers } else { - await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message); + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + + } } } @@ -70,9 +81,9 @@ namespace JiShe.CollectBus.Subscribers /// /// [HttpPost] - [Route("fiveminute/issued-event")] - [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)] - public async Task ScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + [Route("ammeter/fiveminute/issued-event")] + [CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task AmmeterScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) { _logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理"); var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); @@ -82,7 +93,12 @@ namespace JiShe.CollectBus.Subscribers } else { - await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message); + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + + } } } @@ -92,9 +108,9 @@ namespace JiShe.CollectBus.Subscribers /// /// [HttpPost] - [Route("fifteenminute/issued-event")] - [CapSubscribe(ProtocolConst.SubscriberWorkerOneMinuteIssuedEventName)] - public async Task ScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + [Route("ammeter/fifteenminute/issued-event")] + [CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task AmmeterScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) { _logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理"); var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); @@ -104,8 +120,96 @@ namespace JiShe.CollectBus.Subscribers } else { - await _tcpService.SendAsync(receivedMessage.ClientId, receivedMessage.Message); + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + + } } } + #endregion + + #region 水表消息采集 + /// + /// 一分钟定时抄读任务消息消费订阅 + /// + /// + /// + [HttpPost] + [Route("watermeter/oneminute/issued-event")] + [CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task WatermeterScheduledMeterOneMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + { + _logger.LogInformation("1分钟采集电表数据下行消息消费队列开始处理"); + var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); + if (protocolPlugin == null) + { + _logger.LogError("【1分钟采集电表数据下行消息消费队列开始处理】协议不存在！"); + } + else + { + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + + } + } + } + + /// + /// 5分钟采集电表数据下行消息消费订阅 + /// + /// + /// + [HttpPost] + [Route("watermeter/fiveminute/issued-event")] + [CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task WatermeterScheduledMeterFiveMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + { + _logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理"); + var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); + if (protocolPlugin == null) + { + _logger.LogError("【5分钟采集电表数据下行消息消费队列开始处理】协议不存在！"); + } + else + { + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + + } + } + } + + /// + /// 15分钟采集电表数据下行消息消费订阅 + /// + /// + /// + [HttpPost] + [Route("watermeter/fifteenminute/issued-event")] + [CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)] + public async Task WatermeterScheduledMeterFifteenMinuteReadingIssuedEvent(IssuedEventMessage receivedMessage) + { + _logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理"); + var protocolPlugin = _serviceProvider.GetKeyedService("StandardProtocolPlugin"); + if (protocolPlugin == null) + { + _logger.LogError("【15分钟采集电表数据下行消息消费队列开始处理】协议不存在！"); + } + else + { + var device = await _deviceRepository.FindAsync(a => a.Number == receivedMessage.DeviceNo); + if (device != null) + { + await _tcpService.SendAsync(device.ClientId, receivedMessage.Message); + } + } + } + #endregion } } diff --git a/src/JiShe.CollectBus.Common/Consts/FreeRedisConst.cs b/src/JiShe.CollectBus.Common/Consts/FreeRedisConst.cs index ae37f9d..859e86e 100644 --- a/src/JiShe.CollectBus.Common/Consts/FreeRedisConst.cs +++ b/src/JiShe.CollectBus.Common/Consts/FreeRedisConst.cs @@ -13,14 +13,29 @@ namespace JiShe.CollectBus.Common.Consts /// public const string CacheBasicDirectoryKey = "CollectBus:"; + /// + /// 1分钟采集间隔 + /// + public const string OneMinuteAcquisitionTimeInterval = $"one"; + + /// + /// 5分钟采集间隔 + /// + public const string FiveMinuteAcquisitionTimeInterval = $"Five"; + + /// + /// 15分钟采集间隔 + /// + public const string FifteenMinuteAcquisitionTimeInterval = $"Fifteen"; + /// /// 缓存电表信息 /// - public const string CacheAmmeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:AmmeterInfo:"; + public const string CacheAmmeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:{"{1}"}:AmmeterInfo:"; /// /// 缓存水表信息 /// - public const string CacheWatermeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:WatermeterInfo:"; + public const string CacheWatermeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}"}:{"{1}"}:WatermeterInfo:"; } } diff --git a/src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs b/src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs index 699ba43..afca265 100644 --- a/src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs +++ b/src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs @@ -222,7 +222,7 @@ namespace JiShe.CollectBus.Host { context.Services.AddTcpService(config => { - config.SetListenIPHosts(int.Parse(configuration["TCP:ClientPort"] ?? "32580")) + config.SetListenIPHosts(int.Parse(configuration["TCP:ClientPort"] ?? "10500")) //.SetTcpDataHandlingAdapter(()=>new StandardFixedHeaderDataHandlingAdapter()) .SetGetDefaultNewId(() => Guid.NewGuid().ToString())//定义ClinetId的生成策略 .ConfigurePlugins(a => diff --git a/src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs b/src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs index 431fc4c..ee2cddf 100644 --- a/src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs +++ b/src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs @@ -17,15 +17,28 @@ namespace JiShe.CollectBus.Protocol.Contracts /// /// 1分钟采集电表数据下行消息主题 /// - public const string SubscriberWorkerOneMinuteIssuedEventName = "issued.oneminute.event"; + public const string AmmeterSubscriberWorkerOneMinuteIssuedEventName = "issued.one.ammeter.event"; /// /// 5分钟采集电表数据下行消息主题 /// - public const string SubscriberWorkerFiveMinuteIssuedEventName = "issued.fiveminute.event"; + public const string AmmeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.five.ammeter.event"; /// /// 15分钟采集电表数据下行消息主题 /// - public const string SubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteenminute.event"; + public const string AmmeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.ammeter.event"; + + /// + /// 1分钟采集水表数据下行消息主题 + /// + public const string WatermeterSubscriberWorkerOneMinuteIssuedEventName = "issued.one.watermeter.event"; + /// + /// 5分钟采集水表数据下行消息主题 + /// + public const string WatermeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.five.watermeter.event"; + /// + /// 15分钟采集水表数据下行消息主题 + /// + public const string WatermeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.watermeter.event"; } }