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

优化异常捕捉，修复IoTDBEntity基类属性异常的问题

Browse Source
This commit is contained in:
ChenYi 2025-05-14 17:41:39 +08:00
parent ba0af3a12a
commit be3cd5d3e7
12 changed files with 640 additions and 512 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
modules/JiShe.CollectBus.IoTDB/Model/IoTEntity.cs
View File

@ -45,25 +45,27 @@ namespace JiShe.CollectBus.IoTDB.Model
/// </summary>
public long Timestamps { get; set; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
/// <summary>
/// 设备路径
/// </summary>
private string _devicePath;
/// <summary>
/// 设备路径
/// </summary>
public virtual string DevicePath
{
get
{
// 如果未手动设置路径，则自动生成
if (string.IsNullOrWhiteSpace(_devicePath))
{
return $"root.{SystemName.ToLower()}.`{ProjectId}`.`{DeviceType}`.{DataType}.`{DeviceId}`";
}
return _devicePath;
}
set
{
if (string.IsNullOrWhiteSpace(value))
{
DevicePath = $"root.{SystemName.ToLower()}.`{ProjectId}`.`{DeviceType}`.{DataType}.`{DeviceId}`";
}
else
{
DevicePath = value;
}
_devicePath = value; // 直接赋值给支持字段，避免递归
}
}
}

1
modules/JiShe.CollectBus.IoTDB/Options/QueryCondition.cs
View File

@ -56,7 +56,6 @@ namespace JiShe.CollectBus.IoTDB.Options
return declaredTypeName?.ToUpper() switch
{
"DATETIME" => v => v != null ? ((DateTime)v).ToUniversalTime().Ticks : null,
"BOOLEAN" => v => v != null && (bool)v ? 1 : 0,
"STRING" => v => v != null ? $"'{v}'" : "''",
_ => v => v
};

57
modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
View File

@ -1,6 +1,7 @@
using Confluent.Kafka;
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization;
using Microsoft.AspNetCore.DataProtection.KeyManagement;
@ -126,6 +127,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="messageHandler"></param>
/// <returns></returns>
public async Task SubscribeAsync<TKey, TValue>(string[] topics, Func<TKey, TValue, Task<bool>> messageHandler, string? groupId = null) where TKey : notnull where TValue : class
{
try
{
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
{
@ -202,6 +205,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
});
}
catch (Exception ex)
{
throw;
}
}
@ -214,6 +223,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="groupId"></param>
/// <returns></returns>
public async Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId) where TValue : class
{
try
{
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
{
@ -286,6 +297,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.CompletedTask;
});
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
@ -299,9 +316,17 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
public async Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
{
try
{
await SubscribeBatchAsync<TKey, TValue>(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout);
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
/// 批量订阅消息
@ -314,6 +339,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
public async Task SubscribeBatchAsync<TKey, TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
{
try
{
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
{
@ -427,6 +454,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.CompletedTask;
});
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
@ -440,8 +473,16 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchTimeout">批次超时时间</param>
/// <param name="consumeTimeout">消费等待时间</param>
public async Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class
{
try
{
await SubscribeBatchAsync(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout, consumeTimeout);
}
catch (Exception ex)
{
throw;
}
}
@ -457,6 +498,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchTimeout">批次超时时间</param>
/// <param name="consumeTimeout">消费等待时间</param>
public async Task SubscribeBatchAsync<TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class
{
try
{
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
{
@ -569,6 +612,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.CompletedTask;
});
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
@ -577,6 +626,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <typeparam name="TKey"></typeparam>
/// <typeparam name="TValue"></typeparam>
public void Unsubscribe<TKey, TValue>(string[] topics, string? groupId) where TKey : notnull where TValue : class
{
try
{
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
if (_consumerStore.TryRemove(consumerKey, out var entry))
@ -586,6 +637,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
entry.CTS.Dispose();
}
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
/// 释放资源

18
modules/JiShe.CollectBus.Kafka/KafkaSubscribeExtensions.cs
View File

@ -226,6 +226,11 @@ namespace JiShe.CollectBus.Kafka
// 处理消费错误
logger.LogError($"kafka批量消费异常:{ex.Message}");
}
catch (Exception ex)
{
// 处理消费错误
logger.LogError($"kafka批量消费异常:{ex.Message}");
}
return await Task.FromResult(false);
}, attr.GroupId, attr.BatchSize, attr.BatchTimeout);
}
@ -248,6 +253,11 @@ namespace JiShe.CollectBus.Kafka
// 处理消费错误
logger.LogError($"kafka消费异常:{ex.Message}");
}
catch (Exception ex)
{
// 处理消费错误
logger.LogError($"kafka批量消费异常:{ex.Message}");
}
return await Task.FromResult(false);
}, attr.GroupId);
}
@ -259,6 +269,8 @@ namespace JiShe.CollectBus.Kafka
/// 处理消息
/// </summary>
private static async Task<bool> ProcessMessageAsync(List<dynamic> messages, MethodInfo method, object subscribe)
{
try
{
var parameters = method.GetParameters();
bool isGenericTask = method.ReturnType.IsGenericType
@ -380,6 +392,12 @@ namespace JiShe.CollectBus.Kafka
}
return false;
}
catch (Exception ex)
{
throw;
}
}
}

32
modules/JiShe.CollectBus.Kafka/Producer/ProducerService.cs
View File

@ -116,6 +116,8 @@ namespace JiShe.CollectBus.Kafka.Producer
/// <param name="value"></param>
/// <returns></returns>
public async Task ProduceAsync<TKey, TValue>(string topic, TKey key, TValue value)where TKey : notnull where TValue : class
{
try
{
var typeKey = typeof(KafkaProducer<TKey, TValue>);
var producer = GetProducer<TKey, TValue>(typeKey);
@ -129,6 +131,12 @@ namespace JiShe.CollectBus.Kafka.Producer
};
await producer.ProduceAsync(topic, message);
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
/// 发布消息
@ -138,6 +146,8 @@ namespace JiShe.CollectBus.Kafka.Producer
/// <param name="value"></param>
/// <returns></returns>
public async Task ProduceAsync<TValue>(string topic, TValue value) where TValue : class
{
try
{
var typeKey = typeof(KafkaProducer<string, TValue>);
var producer = GetProducer<Null, TValue>(typeKey);
@ -150,6 +160,12 @@ namespace JiShe.CollectBus.Kafka.Producer
};
await producer.ProduceAsync(topic, message);
}
catch (Exception ex)
{
throw;
}
}
/// <summary>
/// 发布消息
@ -163,6 +179,8 @@ namespace JiShe.CollectBus.Kafka.Producer
/// <param name="deliveryHandler"></param>
/// <returns></returns>
public async Task ProduceAsync<TKey, TValue>(string topic,TKey key,TValue value,int? partition=null, Action<DeliveryReport<TKey, TValue>>? deliveryHandler = null)where TKey : notnull where TValue : class
{
try
{
var message = new Message<TKey, TValue>
{
@ -184,6 +202,12 @@ namespace JiShe.CollectBus.Kafka.Producer
producer.Produce(topic, message, deliveryHandler);
}
await Task.CompletedTask;
}
catch (Exception ex)
{
throw;
}
}
@ -198,6 +222,8 @@ namespace JiShe.CollectBus.Kafka.Producer
/// <param name="deliveryHandler"></param>
/// <returns></returns>
public async Task ProduceAsync<TValue>(string topic, TValue value, int? partition=null, Action<DeliveryReport<Null, TValue>>? deliveryHandler = null) where TValue : class
{
try
{
var message = new Message<Null, TValue>
{
@ -220,6 +246,12 @@ namespace JiShe.CollectBus.Kafka.Producer
}
await Task.CompletedTask;
}
catch (Exception ex)
{
throw;
}
}
public void Dispose()
{

18
modules/JiShe.CollectBus.Kafka/Serialization/JsonSerializer.cs
View File

@ -19,6 +19,7 @@ namespace JiShe.CollectBus.Kafka.Serialization
{
DefaultIgnoreCondition = JsonIgnoreCondition.Never,
WriteIndented = false,// 设置格式化输出
IncludeFields = true,// 允许反序列化到非公共 setter 和字段
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,// 允许特殊字符
IgnoreReadOnlyFields = true,
IgnoreReadOnlyProperties = true,
@ -101,11 +102,14 @@ namespace JiShe.CollectBus.Kafka.Serialization
return jsonObject is JsonElement;
}
public static object? Deserialize(object value, Type valueType)
{
try
{
var _jsonSerializerOptions = new JsonSerializerOptions
{
DefaultIgnoreCondition = JsonIgnoreCondition.Never,
WriteIndented = false,// 设置格式化输出
IncludeFields = true,// 允许反序列化到非公共 setter 和字段
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,// 允许特殊字符
IgnoreReadOnlyFields = true,
IgnoreReadOnlyProperties = true,
@ -117,9 +121,19 @@ namespace JiShe.CollectBus.Kafka.Serialization
Converters = { new DateTimeJsonConverter() } // 注册你的自定义转换器,
};
if (value is JsonElement jsonElement) return jsonElement.Deserialize(valueType, _jsonSerializerOptions);
if (value is JsonElement jsonElement)
{
//return jsonElement.Deserialize(valueType, _jsonSerializerOptions);
return JsonSerializer.Deserialize(jsonElement, valueType, _jsonSerializerOptions);
}
throw new NotSupportedException("Type is not of type JsonElement");
return null;
}
catch (Exception ex)
{
throw;
}
}
}
}

4
protocols/JiShe.CollectBus.Protocol.T1882018/SendData/Telemetry1882018PacketBuilder.cs
View File

@ -46,7 +46,7 @@ namespace JiShe.CollectBus.Protocol.T1882018.SendData
{
var itemCodeArr = request.ItemCode.Split('_');
var c_data = itemCodeArr[0];//01
var d_data = itemCodeArr[1];//91 或者 90
var d_data = itemCodeArr[2];//91 或者 90
var dataUnit = new List<string>() { "1F", d_data, "00" };
var dataList = Build188SendData.Build188SendCommand(request.MeterAddress, c_data, dataUnit);
@ -64,7 +64,7 @@ namespace JiShe.CollectBus.Protocol.T1882018.SendData
{
var itemCodeArr = request.ItemCode.Split('_');
var c_data = itemCodeArr[0];//01
var d_data = itemCodeArr[1];//55 或者 99
var d_data = itemCodeArr[2];//55 或者 99
var dataUnit = new List<string>() { "A0", "17", "00", d_data };
var dataList = Build188SendData.Build188SendCommand(request.MeterAddress, c_data, dataUnit);

2
protocols/JiShe.CollectBus.Protocol.T1882018/T1882018ProtocolPlugin.cs
View File

@ -93,6 +93,8 @@ namespace JiShe.CollectBus.Protocol.T1882018
//数据转发场景 10H_F1
if (request.ItemCode == T37612012PacketItemCodeConst.AFN10HFN01H && request.SubProtocolRequest != null && string.IsNullOrWhiteSpace(request.SubProtocolRequest.ItemCode) == false)
{
//var subItemCodeArr = request.SubProtocolRequest.ItemCode.Split("_");
var t188PacketHandlerName = $"{T1882018PacketItemCodeConst.BasicT1882018}_{request.SubProtocolRequest.ItemCode}_Send";
Telemetry1882018PacketResponse t645PacketResponse = null;

4
protocols/JiShe.CollectBus.Protocol.T6452007/SendData/Telemetry6452007PacketBuilder.cs
View File

@ -47,7 +47,7 @@ namespace JiShe.CollectBus.Protocol.T6452007.SendData
{
var itemCodeArr = request.ItemCode.Split('_');
var c_data = itemCodeArr[0];
var n_data = itemCodeArr[1];
var n_data = itemCodeArr[2];
string password = request.Password;
string pwdLevel = "02";
@ -78,7 +78,7 @@ namespace JiShe.CollectBus.Protocol.T6452007.SendData
{
var itemCodeArr = request.ItemCode.Split('_');
var c_data = itemCodeArr[0];
var n_data = itemCodeArr[1];
var n_data = itemCodeArr[2];
string password = request.Password;
if (!string.IsNullOrWhiteSpace(password) && password.Contains("|"))

4
protocols/JiShe.CollectBus.Protocol.T6452007/T6452007ProtocolPlugin.cs
View File

@ -93,7 +93,9 @@ namespace JiShe.CollectBus.Protocol.T6452007
//数据转发场景 10H_F1
if (request.ItemCode == T37612012PacketItemCodeConst.AFN10HFN01H && request.SubProtocolRequest != null && string.IsNullOrWhiteSpace(request.SubProtocolRequest.ItemCode) == false)
{
var t645PacketHandlerName = $"C{request.SubProtocolRequest.ItemCode}_Send";
var subItemCodeArr = request.SubProtocolRequest.ItemCode.Split("_");
var t645PacketHandlerName = $"C{subItemCodeArr[0]}_{subItemCodeArr[1]}_Send";//C1C_01_Send
Telemetry6452007PacketResponse t645PacketResponse = null;
if (T645ControlHandlers != null && T645ControlHandlers.TryGetValue(t645PacketHandlerName

16
services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -122,10 +122,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
TypeName = 1,
DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
TimeDensity = 15,
BrandType = "DDS1980",
BrandType = "DTS1980",
MeterType = MeterTypeEnum.Ammeter,
ProjectID = 1,
MeteringPort = 2,
Password = "000000",
});
ammeterInfos.Add(new DeviceInfo()
@ -141,10 +142,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
TypeName = 1,
DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
TimeDensity = 15,
BrandType = "DDS1980",
BrandType = "DTS1980",
MeterType = MeterTypeEnum.Ammeter,
ProjectID = 1,
MeteringPort = 2,
Password = "000000",
});
//ammeterInfos.Add(new DeviceInfo()
@ -287,7 +289,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = "442400040",
FocusId = 95780,
ProjectID = 1,
TripType = "off",
TripType = "on",
TripTime = $"{DateTime.Now:HH:mm}",
MeterId = 127035,
LoopType = "EachDay",
@ -301,7 +303,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = "442400039",
FocusId = 69280,
ProjectID = 1,
TripType = "off",
TripType = "on",
TripTime = $"{DateTime.Now:HH:mm}",
MeterId = 95594,
LoopType = "EachDay",
@ -377,17 +379,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
ammeterInfo.TripState = 0;
tripStateResult = true;
subItemCode = T6452007PacketItemCodeConst.C1C01C;
subItemCode = T6452007PacketItemCodeConst.C1C011C;
if (ammeterInfo.TypeName != 1)
{
subItemCode = T6452007PacketItemCodeConst.C1C01B;
subItemCode = T6452007PacketItemCodeConst.C1C011B;
}
}
else if (settingInfo.TripType.Equals("off"))
{
ammeterInfo.TripState = 1;
tripStateResult = false;
subItemCode = T6452007PacketItemCodeConst.C1C01A;
subItemCode = T6452007PacketItemCodeConst.C1C011A;
}
else
{

14
shared/JiShe.CollectBus.Common/Consts/T6452007PacketItemCodeConst.cs
View File

@ -16,37 +16,37 @@ namespace JiShe.CollectBus.Common.Consts
/// <summary>
/// 跳闸
/// </summary>
public const string C1C01A = "1C_1A";
public const string C1C011A = "1C_01_1A";
/// <summary>
/// 单相合闸
/// </summary>
public const string C1C01B = "1C_1B";
public const string C1C011B = "1C_01_1B";
/// <summary>
/// 三相合闸
/// </summary>
public const string C1C01C = "1C_1C";
public const string C1C011C = "1C_01_1C";
/// <summary>
/// 触发报警
/// </summary>
public const string C1C02A = "1C_2A";
public const string C1C021A = "1C_01_2A";
/// <summary>
/// 报警解除
/// </summary>
public const string C1C02B = "1C_2B";
public const string C1C012B = "1C_01_2B";
/// <summary>
/// 保电开始
/// </summary>
public const string C1C03A = "1C_3A";
public const string C1C033A = "1C_03_3A";
/// <summary>
/// 保电结束
/// </summary>
public const string C1C03B = "1C_3B";
public const string C1C033B = "1C_03_3B";
#endregion
#region 广播校时