admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

合并

Browse Source
This commit is contained in:
cli 2025-03-14 14:38:01 +08:00
commit d82e646359
14 changed files with 852 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/JiShe.CollectBus.Application.Contracts/JiShe.CollectBus.Application.Contracts.csproj
View File

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

10
src/JiShe.CollectBus.Application.Contracts/Workers/IScheduledMeterReadingService.cs → src/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs
View File

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

0
src/JiShe.CollectBus.Application.Contracts/Workers/IWorkerSubscriberAppService.cs → src/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
View File

4
src/JiShe.CollectBus.Application/Consumers/IssuedConsumer.cs
View File

@ -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)

40
src/JiShe.CollectBus.Application/Consumers/WorkerConsumer.cs Normal file
View File

@ -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);
}
}
}

672
src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs Normal file
View File

@ -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
}
}

21
src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs → src/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -1,10 +1,12 @@
using DotNetCore.CAP;
using FreeRedis;
using FreeSql;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.FreeRedisProvider;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.GatherItem;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
@ -33,6 +35,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>

0
src/JiShe.CollectBus.Application/Workers/WorkerSubscriberAppService.cs → src/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
View File

242
src/JiShe.CollectBus.Application/Workers/BasicScheduledMeterReadingService.cs
View File

@ -1,242 +0,0 @@
using FreeRedis;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.IotSystems.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;
public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger)
{
_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(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<string, Dictionary<string, string>>();
if (oneMinuteAmmerterResult is object[] arr)
{
foreach (object[] item in arr)
{
string key = (string)item[0];
object[] fieldsAndValues = (object[])item[1];
var dict = new Dictionary<string, string>();
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分钟采集电表数据处理完成");
}
/// <summary>
/// 5分钟采集电表数据
/// </summary>
/// <returns></returns>
public virtual Task AmmeterScheduledMeterFiveMinuteReading()
{
throw new NotImplementedException($"{nameof(AmmeterScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现");
}
/// <summary>
/// 15分钟采集电表数据
/// </summary>
/// <returns></returns>
public virtual Task AmmeterScheduledMeterFifteenMinuteReading()
{
throw new NotImplementedException($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现");
}
#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.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*";
var oneMinuteList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
_logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成");
}
/// <summary>
/// 5分钟采集电表数据
/// </summary>
/// <returns></returns>
public virtual Task WatermeterScheduledMeterFiveMinuteReading()
{
throw new NotImplementedException($"{nameof(WatermeterScheduledMeterFiveMinuteReading)}请根据不同系统类型进行实现");
}
/// <summary>
/// 15分钟采集电表数据
/// </summary>
/// <returns></returns>
public virtual Task WatermeterScheduledMeterFifteenMinuteReading()
{
throw new NotImplementedException($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)}请根据不同系统类型进行实现");
}
#endregion
}
}

54
src/JiShe.CollectBus.Common/Enums/MeterLinkProtocolEnum.cs Normal file
View File

@ -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,
}
}

2
src/JiShe.CollectBus.Domain/Ammeters/AmmeterInfo.cs
View File

@ -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; }

21
src/JiShe.CollectBus.Domain/GatherItem/GatherItemInfo.cs Normal file
View File

@ -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; }
}
}

2
src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs
View File

@ -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>();

26
src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs
View File

@ -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
}
}