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

Merge branch 'dev' of https://310.jisheyun.com/daizan/JiShe.CollectBus into dev

Browse Source
This commit is contained in:
cli 2025-03-17 15:51:43 +08:00
commit 798c48650a
15 changed files with 2161 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
View File

@ -1,5 +1,6 @@
using System.Threading.Tasks;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using System.Threading.Tasks;
using Volo.Abp.Application.Services;
namespace JiShe.CollectBus.Subscribers

72
src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
@ -13,6 +14,8 @@ using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Protocol.Contracts;
@ -20,6 +23,7 @@ using JiShe.CollectBus.Workers;
using MassTransit;
using MassTransit.Internals.GraphValidation;
using Microsoft.Extensions.Logging;
using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.ScheduledMeterReading
{
@ -30,10 +34,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
private readonly ILogger<BasicScheduledMeterReadingService> _logger;
private readonly ICapPublisher _capBus;
public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger, ICapPublisher capBus)
private readonly IRepository<ScheduledMeterReadingIssuedEventMessage, Guid> _meterReadingIssuedRepository;
public BasicScheduledMeterReadingService(
ILogger<BasicScheduledMeterReadingService> logger,
ICapPublisher capBus,
IRepository<ScheduledMeterReadingIssuedEventMessage, Guid> meterReadingIssuedRepository)
{
_capBus = capBus;
_logger = logger;
_meterReadingIssuedRepository = meterReadingIssuedRepository;
}
/// <summary>
@ -118,7 +129,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
List<string> itemCodeList = new List<string>();
foreach (var dataType in itemArr)
{
var excludeItemCode = "10_98,10_94";//排除透明转发:尖峰平谷时段、跳合闸
var excludeItemCode = "10_98,10_94";//TODO 排除透明转发:尖峰平谷时段、跳合闸,特殊电表
var gatherItem = gatherItemInfos.FirstOrDefault(f => f.DataType.Equals(dataType));
if (gatherItem != null)
{
@ -127,17 +138,21 @@ namespace JiShe.CollectBus.ScheduledMeterReading
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,";
if (itemArr.Exists(e => e.Equals("109")))//WAVE_109
{
itemCodeList.Add("10_109");
}
#endregion
}
#endregion
ammeter.ItemCodes = itemCodeList.Serialize();//转换成JSON字符串
@ -207,7 +222,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-102");
return;
}
await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, meterInfos);
await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName, meterInfos);
_logger.LogInformation($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理完成");
}
@ -235,8 +250,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading
return;
}
await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, meterInfos);
_logger.LogInformation($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理完成");
Stopwatch stopwatch = new Stopwatch();
stopwatch.Start();
await AmmerterScheduledMeterReadingIssued(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName, meterInfos);
stopwatch.Stop();
_logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理完成，共消耗{stopwatch.ElapsedMilliseconds}毫秒。");
}
#endregion
@ -459,7 +480,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
try
{
//将采集器编号的hash值取模分组
const int TotalShards = 20;
const int TotalShards = 1024;
var focusHashGroups = new Dictionary<int, Dictionary<string, Dictionary<string, AmmeterInfo>>>();
foreach (var (collectorId, ammetersDictionary) in focusGroup)
@ -493,10 +514,12 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//根据分组创建线程批处理集中器
foreach (var group in focusHashGroups)
{
_ = Task.Run(async () => { await CreatePublishTask(eventName, group.Value); });
//TODO _meterReadingIssuedRepository 需要优化
//_ = Task.Run(async () => { await CreatePublishTask(eventName, group.Value); });
await CreatePublishTask(eventName, group.Value);
}
await Task.CompletedTask;
//await Task.CompletedTask;
}
catch (Exception)
{
@ -538,12 +561,12 @@ namespace JiShe.CollectBus.ScheduledMeterReading
continue;
}
//排除1天未在线的集中器生成指令 或 排除集中器配置为自动上报的集中器
if (!IsGennerateCmd(ammeter.LastTime, -1))
{
_logger.LogInformation($"{nameof(CreatePublishTask)} 集中器{ammeter.FocusAddress}的电表{ammeter.Name}，采集时间：{ammeter.LastTime}，已超过1天未在线，不生成指令");
continue;
}
////排除1天未在线的集中器生成指令 或 排除集中器配置为自动上报的集中器
//if (!IsGennerateCmd(ammeter.LastTime, -1))
//{
// _logger.LogInformation($"{nameof(CreatePublishTask)} 集中器{ammeter.FocusAddress}的电表{ammeter.Name}，采集时间：{ammeter.LastTime}，已超过1天未在线，不生成指令");
// continue;
//}
if (string.IsNullOrWhiteSpace(ammeter.AreaCode))
{
@ -598,6 +621,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
}
}
List<ScheduledMeterReadingIssuedEventMessage> evenMessageInfoList = new List<ScheduledMeterReadingIssuedEventMessage>();
foreach (var tempItem in tempCodes)
{
//排除已发送日冻结和月冻结采集项配置
@ -622,13 +646,15 @@ namespace JiShe.CollectBus.ScheduledMeterReading
}
else
{
//特殊表暂不处理
if (TelemetryPacketBuilder.AFNHandlers.TryGetValue(tempItem, out var handler))
{
dataInfos = handler(ammeter.FocusAddress, fn, ammeter.MeteringCode);
}
else
{
throw new InvalidOperationException($"无效编码: {tempItem}");
_logger.LogWarning($"{nameof(CreatePublishTask)} 集中器{ammeter.FocusAddress}的电表{ammeter.Name}采集项{tempItem}无效编码。");
continue;
}
}
//TODO:特殊表
@ -643,13 +669,15 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
Message = dataInfos!,
DeviceNo = ammeter.FocusAddress,
MessageId = NewId.NextGuid().ToString()
MessageId = NewId.NextGuid().ToString(),
TimeDensity = eventName,
WasSuccessful = false,
};
await _capBus.PublishAsync(eventName, evenMessageInfo);
evenMessageInfoList.Add(evenMessageInfo);
}
await _meterReadingIssuedRepository.InsertManyAsync(evenMessageInfoList);
}
}
}

12
src/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -1,13 +1,16 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using DotNetCore.CAP;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.Watermeter;
using Microsoft.AspNetCore.Authorization;
using Microsoft.Extensions.Logging;
using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.ScheduledMeterReading
{
@ -19,7 +22,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
{
public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger, ICapPublisher capBus) :base(logger, capBus)
public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger, ICapPublisher capBus, IRepository<ScheduledMeterReadingIssuedEventMessage, Guid> meterReadingIssuedRepository) :base(logger, capBus, meterReadingIssuedRepository)
{
}
@ -59,11 +62,12 @@ namespace JiShe.CollectBus.ScheduledMeterReading
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 ";
WHERE 1=1 and C.Special = 0 ";
//TODO 记得移除特殊表过滤
if (!string.IsNullOrWhiteSpace(gatherCode))
{
sql = $@"{sql} A.GatherCode = '{gatherCode}'";
sql = $@"{sql} AND A.GatherCode = '{gatherCode}'";
}
return await SqlProvider.Instance.Change(DbEnum.EnergyDB)
.Ado

11
src/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
View File

@ -3,8 +3,9 @@ using System.Threading.Tasks;
using DeviceDetectorNET.Parser.Device;
using DotNetCore.CAP;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using Microsoft.AspNetCore.Mvc;
@ -81,7 +82,7 @@ namespace JiShe.CollectBus.Subscribers
/// <returns></returns>
[HttpPost]
[Route("ammeter/fiveminute/issued-event")]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName)]
public async Task AmmeterScheduledMeterFiveMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理");
@ -108,7 +109,7 @@ namespace JiShe.CollectBus.Subscribers
/// <returns></returns>
[HttpPost]
[Route("ammeter/fifteenminute/issued-event")]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
public async Task AmmeterScheduledMeterFifteenMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");
@ -164,7 +165,7 @@ namespace JiShe.CollectBus.Subscribers
/// <returns></returns>
[HttpPost]
[Route("watermeter/fiveminute/issued-event")]
[CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName)]
public async Task WatermeterScheduledMeterFiveMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理");
@ -191,7 +192,7 @@ namespace JiShe.CollectBus.Subscribers
/// <returns></returns>
[HttpPost]
[Route("watermeter/fifteenminute/issued-event")]
[CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
public async Task WatermeterScheduledMeterFifteenMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");

44
src/JiShe.CollectBus.Application/Workers/SubscriberFifteenMinuteWorker.cs Normal file
View File

@ -0,0 +1,44 @@
using System.Threading;
using System.Threading.Tasks;
using Hangfire;
using JiShe.CollectBus.ScheduledMeterReading;
using Microsoft.Extensions.Logging;
using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Uow;
namespace JiShe.CollectBus.Workers
{
/// <summary>
/// 15分钟采集数据
/// </summary>
public class SubscriberFifteenMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency, ICollectWorker
{
private readonly ILogger<SubscriberFifteenMinuteWorker> _logger;
private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberFifteenMinuteWorker"/> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="scheduledMeterReadingService">定时任务</param>
public SubscriberFifteenMinuteWorker(ILogger<SubscriberFifteenMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
{
_logger = logger;
RecurringJobId = nameof(SubscriberFifteenMinuteWorker);
CronExpression = Cron.Hourly(15);
this._scheduledMeterReadingService = scheduledMeterReadingService;
}
public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
{
await _scheduledMeterReadingService.AmmeterScheduledMeterFifteenMinuteReading();
//using (var uow = LazyServiceProvider.LazyGetRequiredService<IUnitOfWorkManager>().Begin())
//{
// Logger.LogInformation("Executed MyLogWorker..!");
// return Task.CompletedTask;
//}
}
}
}

39
src/JiShe.CollectBus.Application/Workers/SubscriberFiveMinuteWorker.cs Normal file
View File

@ -0,0 +1,39 @@
using System.Threading;
using System.Threading.Tasks;
using Hangfire;
using JiShe.CollectBus.ScheduledMeterReading;
using Microsoft.Extensions.Logging;
using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Uow;
namespace JiShe.CollectBus.Workers
{
/// <summary>
/// 5分钟采集数据
/// </summary>
public class SubscriberFiveMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
{
private readonly ILogger<SubscriberFiveMinuteWorker> _logger;
private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberFifteenMinuteWorker"/> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="scheduledMeterReadingService">定时任务</param>
public SubscriberFiveMinuteWorker(ILogger<SubscriberFiveMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
{
_logger = logger;
RecurringJobId = nameof(SubscriberFiveMinuteWorker);
CronExpression = Cron.Hourly(15);
this._scheduledMeterReadingService = scheduledMeterReadingService;
}
public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
{
await _scheduledMeterReadingService.AmmeterScheduledMeterFifteenMinuteReading();
}
}
}

39
src/JiShe.CollectBus.Application/Workers/SubscriberOneMinuteWorker.cs Normal file
View File

@ -0,0 +1,39 @@
using System.Threading;
using System.Threading.Tasks;
using Hangfire;
using JiShe.CollectBus.ScheduledMeterReading;
using Microsoft.Extensions.Logging;
using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Uow;
namespace JiShe.CollectBus.Workers
{
/// <summary>
/// 1分钟采集数据
/// </summary>
public class SubscriberOneMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
{
private readonly ILogger<SubscriberOneMinuteWorker> _logger;
private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberFifteenMinuteWorker"/> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="scheduledMeterReadingService">定时任务</param>
public SubscriberOneMinuteWorker(ILogger<SubscriberOneMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
{
_logger = logger;
RecurringJobId = nameof(SubscriberOneMinuteWorker);
CronExpression = Cron.Hourly(15);
this._scheduledMeterReadingService = scheduledMeterReadingService;
}
public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
{
await _scheduledMeterReadingService.AmmeterScheduledMeterFifteenMinuteReading();
}
}
}

2
src/JiShe.CollectBus.Common/BuildSendDatas/Build3761SendData.cs
View File

@ -1401,7 +1401,7 @@ namespace JiShe.CollectBus.Common.BuildSendDatas
cmdStrList.AddRange(userDatas);
cmdStrList.Add(cs);
cmdStrList.Add(endStr);
Console.WriteLine(string.Join(" ", cmdStrList));
//Console.WriteLine(string.Join(" ", cmdStrList));
var bytes = cmdStrList.Select(x => Convert.ToByte(x, 16)).ToArray();
return bytes;
}

1936
src/JiShe.CollectBus.Common/BuildSendDatas/TelemetryPacketBuilder.cs
View File

File diff suppressed because it is too large Load Diff

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

@ -119,7 +119,7 @@ namespace JiShe.CollectBus.Ammeters
public string GatherCode { get; set; }
/// <summary>
/// 是否特殊表
/// 是否特殊表，1是特殊电表
/// </summary>
public int Special { get; set; }

24
src/JiShe.CollectBus.Common/Models/ScheduledMeterReadingIssuedEventMessage.cs → src/JiShe.CollectBus.Domain/IotSystems/MessageIssueds/ScheduledMeterReadingIssuedEventMessage.cs
View File

@ -1,17 +1,14 @@
using JiShe.CollectBus.Common.Enums;
using System;
using Volo.Abp.Domain.Entities;
namespace JiShe.CollectBus.Common.Models
namespace JiShe.CollectBus.IotSystems.MessageIssueds
{
/// <summary>
/// 定时抄读Kafka消息实体，1分钟、5分钟、15分钟
/// </summary>
public class ScheduledMeterReadingIssuedEventMessage
public class ScheduledMeterReadingIssuedEventMessage : AggregateRoot<Guid>
{
/// <summary>
/// 消息接收客户端Id
/// </summary>
public string ClientId { get; set; }
/// <summary>
/// 消息内容
/// </summary>
@ -22,14 +19,19 @@ namespace JiShe.CollectBus.Common.Models
/// </summary>
public string DeviceNo { get; set; }
///// <summary>
///// 采集时间间隔(分钟，如15)
///// </summary>
//public int TimeDensity { get; set; }
/// <summary>
/// 采集时间间隔，通过Kafka主题区分(分钟，如15)
/// </summary>
public string TimeDensity { get; set; }
/// <summary>
/// 消息Id
/// </summary>
public string MessageId { get; set; }
/// <summary>
/// 是否下发成功
/// </summary>
public bool WasSuccessful { get; set; }
}
}

2
src/JiShe.CollectBus.Host/appsettings.Development.json
View File

@ -1,7 +1,7 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Default": "Warning",
"Microsoft.AspNetCore": "Warning"
}
}

2
src/JiShe.CollectBus.Host/appsettings.json
View File

@ -5,7 +5,7 @@
"Serilog.Sinks.File"
],
"MinimumLevel": {
"Default": "Information",
"Default": "Warning",
"Override": {
"Microsoft": "Information",
"Volo.Abp": "Warning",

3
src/JiShe.CollectBus.MongoDB/MongoDB/CollectBusMongoDbContext.cs
View File

@ -1,4 +1,5 @@
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.Protocols;
using MongoDB.Driver;
@ -22,6 +23,8 @@ public class CollectBusMongoDbContext : AbpMongoDbContext, ICollectBusMongoDbCon
public IMongoCollection<Device> Devices => Collection<Device>();
public IMongoCollection<ProtocolInfo> ProtocolInfos => Collection<ProtocolInfo>();
public IMongoCollection<ScheduledMeterReadingIssuedEventMessage> MeterReadingIssued => Collection<ScheduledMeterReadingIssuedEventMessage>();
protected override void CreateModel(IMongoModelBuilder modelBuilder)
{
base.CreateModel(modelBuilder);