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

封装水电表redis缓存批量获取

Browse Source
This commit is contained in:
ChenYi 2025-03-13 11:55:51 +08:00
parent 0b0eaa6ad0
commit addec01dc8
7 changed files with 389 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,6 +1,8 @@
using FreeRedis; using DotNetCore.CAP;
using FreeRedis;
using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Watermeter; using JiShe.CollectBus.Watermeter;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using System; using System;
@ -19,8 +21,10 @@ namespace JiShe.CollectBus.Workers
public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
{ {
private readonly ILogger<BasicScheduledMeterReadingService> _logger; private readonly ILogger<BasicScheduledMeterReadingService> _logger;
public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger) private readonly ICapPublisher _capBus;
public BasicScheduledMeterReadingService(ILogger<BasicScheduledMeterReadingService> logger, ICapPublisher capBus)
{ {
_capBus = capBus;
_logger = logger; _logger = logger;
} }
@ -90,48 +94,21 @@ namespace JiShe.CollectBus.Workers
//获取缓存中的电表信息 //获取缓存中的电表信息
var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*"; var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*";
var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList); var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
if (oneMinutekeyList == null || oneMinutekeyList.Length <=0) if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
{ {
_logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102"); _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-101");
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; return;
} }
// 解析结果（结果为嵌套数组） // 解析结果（结果为嵌套数组）
var parsedResults = new Dictionary<string, Dictionary<string, string>>(); List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
if (oneMinuteAmmerterResult is object[] arr) if (meterInfos == null || meterInfos.Count <= 0)
{ {
foreach (object[] item in arr) _logger.LogError($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理时没有获取到缓存信息,-102");
{ return;
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分钟采集电表数据处理完成"); _logger.LogInformation($"{nameof(AmmeterScheduledMeterOneMinuteReading)} 1分钟采集电表数据处理完成");
} }
@ -139,18 +116,52 @@ namespace JiShe.CollectBus.Workers
/// 5分钟采集电表数据 /// 5分钟采集电表数据
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
public virtual Task AmmeterScheduledMeterFiveMinuteReading() public virtual async Task AmmeterScheduledMeterFiveMinuteReading()
{ {
throw new NotImplementedException($"{nameof(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;
}
// 解析结果（结果为嵌套数组）
List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理时没有获取到缓存信息,-102");
return;
}
_logger.LogInformation($"{nameof(AmmeterScheduledMeterFiveMinuteReading)} 5分钟采集电表数据处理完成");
} }
/// <summary> /// <summary>
/// 15分钟采集电表数据 /// 15分钟采集电表数据
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
public virtual Task AmmeterScheduledMeterFifteenMinuteReading() public virtual async Task AmmeterScheduledMeterFifteenMinuteReading()
{ {
throw new NotImplementedException($"{nameof(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;
}
// 解析结果（结果为嵌套数组）
List<AmmeterInfo> meterInfos = await GetMeterCacheData<AmmeterInfo>(oneMinutekeyList);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理时没有获取到缓存信息,-102");
return;
}
_logger.LogInformation($"{nameof(AmmeterScheduledMeterFifteenMinuteReading)} 15分钟采集电表数据处理完成");
} }
#endregion #endregion
@ -211,9 +222,22 @@ namespace JiShe.CollectBus.Workers
/// <returns></returns> /// <returns></returns>
public virtual async Task WatermeterScheduledMeterOneMinuteReading() public virtual async Task WatermeterScheduledMeterOneMinuteReading()
{ {
//获取缓存中的电表信息 //获取缓存中的水表信息
var redisKeyList = $"{string.Format(FreeRedisConst.CacheAmmeterInfoKey, SystemTypeConst.Energy, 1)}*"; var redisKeyList = $"{string.Format(FreeRedisConst.CacheWatermeterInfoKey, SystemTypeConst.Energy, 1)}*";
var oneMinuteList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList); var oneMinutekeyList = await FreeRedisProvider.FreeRedis.KeysAsync(redisKeyList);
if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
{
_logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表据处理时没有获取到缓存信息,-101");
return;
}
// 解析结果（结果为嵌套数组）
List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理时没有获取到缓存信息,-102");
return;
}
_logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成"); _logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集水表数据处理完成");
} }
@ -222,21 +246,113 @@ namespace JiShe.CollectBus.Workers
/// 5分钟采集电表数据 /// 5分钟采集电表数据
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
public virtual Task WatermeterScheduledMeterFiveMinuteReading() public virtual async Task WatermeterScheduledMeterFiveMinuteReading()
{ {
throw new NotImplementedException($"{nameof(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;
}
// 解析结果（结果为嵌套数组）
List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理时没有获取到缓存信息,-102");
return;
}
_logger.LogInformation($"{nameof(WatermeterScheduledMeterFiveMinuteReading)} 5分钟采集水表数据处理完成");
} }
/// <summary> /// <summary>
/// 15分钟采集电表数据 /// 15分钟采集电表数据
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
public virtual Task WatermeterScheduledMeterFifteenMinuteReading() public virtual async Task WatermeterScheduledMeterFifteenMinuteReading()
{ {
throw new NotImplementedException($"{nameof(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;
}
// 解析结果（结果为嵌套数组）
List<WatermeterInfo> meterInfos = await GetMeterCacheData<WatermeterInfo>(oneMinutekeyList);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理时没有获取到缓存信息,-102");
return;
}
_logger.LogInformation($"{nameof(WatermeterScheduledMeterFifteenMinuteReading)} 15分钟采集水表数据处理完成");
} }
#endregion #endregion
#region 公共处理方法
/// <summary>
/// 批量获取缓存的表计信息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisKeys"></param>
/// <returns></returns>
private async Task<List<T>> GetMeterCacheData<T>(string[] redisKeys)
{
//通过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)} 1分钟采集电表数据处理时没有获取到缓存信息,-102");
return null;
}
// 解析结果（结果为嵌套数组）
List<T> meterInfos = new List<T>();
if (oneMinuteAmmerterResult is object[] arr)
{
foreach (object[] item in arr)
{
string key = (string)item[0];
object[] fieldsAndValues = (object[])item[1];
for (int i = 0; i < fieldsAndValues.Length; i += 2)
{
string field = (string)fieldsAndValues[i];
string valueStr = (string)fieldsAndValues[i + 1];
T value = default;
if (!string.IsNullOrWhiteSpace(valueStr))
{
value = valueStr.Deserialize<T>()!;
}
if (value != null)
{
meterInfos.Add(value);
}
else
{
_logger.LogInformation($"{nameof(WatermeterScheduledMeterOneMinuteReading)} 1分钟采集电表{key}数据{field}处理异常");
}
}
}
}
return meterInfos;
}
#endregion
} }
} }

10
src/JiShe.CollectBus.Application/Workers/EnergySystemScheduledMeterReadingService.cs
View File

@ -1,4 +1,5 @@
using FreeRedis; using DotNetCore.CAP;
using FreeRedis;
using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Devices; using JiShe.CollectBus.Devices;
@ -25,13 +26,10 @@ namespace JiShe.CollectBus.Workers
//[Route($"/energy/app/scheduled")] //[Route($"/energy/app/scheduled")]
public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
{ {
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly ILogger<BasicScheduledMeterReadingService> _logger;
public EnergySystemScheduledMeterReadingService(IRepository<Device, Guid> deviceRepository, ILogger<BasicScheduledMeterReadingService> logger):base(logger) public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger, ICapPublisher capBus) :base(logger, capBus)
{ {
this._deviceRepository = deviceRepository;
this._logger = logger;
} }
public sealed override string SystemType => SystemTypeConst.Energy; public sealed override string SystemType => SystemTypeConst.Energy;

126
src/JiShe.CollectBus.Common/Helpers/JsonHelper.cs Normal file
View File

@ -0,0 +1,126 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Encodings.Web;
using System.Text.Json.Serialization;
using System.Text.Json;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Common.Helpers
{
/// <summary>
/// json帮助类
/// </summary>
public static class JsonHelper
{
/// <summary>
/// json对象转换成字符串
/// </summary>
/// <param name="obj">需要序列化的对象</param>
/// <param name="IsIgnore">是否忽略实体中实体，不再序列化里面包含的实体</param>
/// <param name="jsonSerializerOptions">配置</param>
/// <returns></returns>
public static string Serialize(this object obj, bool IsIgnore = false, JsonSerializerOptions jsonSerializerOptions = null)
{
if (jsonSerializerOptions == null)
{
jsonSerializerOptions = new JsonSerializerOptions
{
DefaultIgnoreCondition = JsonIgnoreCondition.Never,
WriteIndented = false,
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
};
}
if (jsonSerializerOptions.Converters != null)
{
jsonSerializerOptions.Converters.Add(new DateTimeJsonConverter());
}
if (IsIgnore == true)
{
jsonSerializerOptions.ReferenceHandler = ReferenceHandler.IgnoreCycles; // 忽略循环引用
return JsonSerializer.Serialize(obj, jsonSerializerOptions);
}
else
{
return JsonSerializer.Serialize(obj, jsonSerializerOptions);
}
}
/// <summary>
/// json字符串转换成json对象
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="json"></param>
/// <param name="IsIgnore">是否忽略实体中实体，不再序列化里面包含的实体</param>
/// <param name="jsonSerializerOptions">配置</param>
/// <returns></returns>
public static T? Deserialize<T>(this string json, bool IsIgnore = false, JsonSerializerOptions jsonSerializerOptions = null)
{
if (jsonSerializerOptions == null)
{
jsonSerializerOptions = new JsonSerializerOptions
{
DefaultIgnoreCondition = JsonIgnoreCondition.Never,
WriteIndented = false,
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
};
}
if (jsonSerializerOptions.Converters != null)
{
jsonSerializerOptions.Converters.Add(new DateTimeJsonConverter());
}
if (IsIgnore == true)
{
jsonSerializerOptions.ReferenceHandler = ReferenceHandler.IgnoreCycles; // 忽略循环引用
return json == null ? default(T) : JsonSerializer.Deserialize<T>(json, jsonSerializerOptions);
}
else
{
return json == null ? default(T) : JsonSerializer.Deserialize<T>(json, jsonSerializerOptions);
}
}
/// <summary>
/// list json字符串转换成list
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="json"></param>
/// <returns></returns>
public static List<T>? DeserializeToList<T>(this string json)
{
return json == null ? default(List<T>) : Deserialize<List<T>>(json);
}
}
public class DateTimeJsonConverter : JsonConverter<DateTime>
{
private readonly string _dateFormatString;
public DateTimeJsonConverter()
{
_dateFormatString = "yyyy-MM-dd HH:mm:ss";
}
public DateTimeJsonConverter(string dateFormatString)
{
_dateFormatString = dateFormatString;
}
public override DateTime Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
{
return DateTime.Parse(reader.GetString());
}
public override void Write(Utf8JsonWriter writer, DateTime value, JsonSerializerOptions options)
{
writer.WriteStringValue(value.ToString(_dateFormatString));
}
}
}

9
src/JiShe.CollectBus.FreeRedisProvider/FreeRedisProviderModule.cs
View File

@ -8,10 +8,10 @@ using System.Text;
using System.Threading.Tasks; using System.Threading.Tasks;
using Volo.Abp.Modularity; using Volo.Abp.Modularity;
namespace JiShe.CollectBus.FreeRedisProvider; namespace JiShe.CollectBus.FreeRedisProvider
public class FreeRedisProviderModule : AbpModule
{ {
public class FreeRedisProviderModule : AbpModule
{
public override void ConfigureServices(ServiceConfigurationContext context) public override void ConfigureServices(ServiceConfigurationContext context)
{ {
var configuration = context.Services.GetConfiguration(); var configuration = context.Services.GetConfiguration();
@ -22,6 +22,9 @@ public class FreeRedisProviderModule : AbpModule
}); });
}
} }
} }

24
src/JiShe.CollectBus.FreeRedisProvider/FreeRedisProviderService.cs
View File

@ -11,10 +11,11 @@ using System.Text.Json;
using System.Threading.Tasks; using System.Threading.Tasks;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
namespace JiShe.CollectBus.FreeRedisProvider; namespace JiShe.CollectBus.FreeRedisProvider
public class FreeRedisProviderService : IFreeRedisProviderService, ISingletonDependency
{ {
public class FreeRedisProviderService : IFreeRedisProviderService, ISingletonDependency
{
private FreeRedisOptions _freeRedisOptions; private FreeRedisOptions _freeRedisOptions;
/// <summary> /// <summary>
@ -41,12 +42,12 @@ public class FreeRedisProviderService : IFreeRedisProviderService, ISingletonDep
public IRedisClient GetClient() public IRedisClient GetClient()
{ {
string connectionString = $"{_freeRedisOptions.Configuration},defaultdatabase={_freeRedisOptions.DefaultDB}"; string connectionString = $"{_freeRedisOptions.Configuration},defaultdatabase={_freeRedisOptions.DefaultDB}";
var redisClinet = new RedisClient(connectionString); var redisClient = new RedisClient(connectionString);
redisClinet.Serialize = obj => JsonSerializer.Serialize(obj); redisClient.Serialize = obj => JsonSerializer.Serialize(obj);
redisClinet.Deserialize = (json, type) => JsonSerializer.Deserialize(json, type); redisClient.Deserialize = (json, type) => JsonSerializer.Deserialize(json, type);
redisClinet.Notice += (s, e) => Trace.WriteLine(e.Log); redisClient.Notice += (s, e) => Trace.WriteLine(e.Log);
return redisClinet; return redisClient;
} }
/// <summary> /// <summary>
@ -56,8 +57,9 @@ public class FreeRedisProviderService : IFreeRedisProviderService, ISingletonDep
/// <returns></returns> /// <returns></returns>
public IRedisClient GetDatabase(int index = 0) public IRedisClient GetDatabase(int index = 0)
{ {
var redisClinet = GetClient(); var redisClient = GetClient();
redisClinet.GetDatabase(index); redisClient.GetDatabase(index);
return redisClinet; return redisClient;
}
} }
} }

7
src/JiShe.CollectBus.FreeRedisProvider/IFreeRedisProviderService.cs
View File

@ -5,10 +5,10 @@ using System.Linq;
using System.Text; using System.Text;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace JiShe.CollectBus.FreeRedisProvider; namespace JiShe.CollectBus.FreeRedisProvider
public interface IFreeRedisProviderService
{ {
public interface IFreeRedisProviderService
{
/// <summary> /// <summary>
/// 默认客户端 /// 默认客户端
/// </summary> /// </summary>
@ -26,4 +26,5 @@ public interface IFreeRedisProviderService
/// <param name="index"></param> /// <param name="index"></param>
/// <returns></returns> /// <returns></returns>
IRedisClient GetDatabase(int index = 0); IRedisClient GetDatabase(int index = 0);
}
} }

8
src/JiShe.CollectBus.FreeRedisProvider/Options/FreeRedisOptions.cs
View File

@ -4,10 +4,10 @@ using System.Linq;
using System.Text; using System.Text;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace JiShe.CollectBus.FreeRedisProvider.Options; namespace JiShe.CollectBus.FreeRedisProvider.Options
public class FreeRedisOptions
{ {
public class FreeRedisOptions
{
/// <summary> /// <summary>
/// 连接字符串 /// 连接字符串
/// </summary> /// </summary>
@ -22,4 +22,6 @@ public class FreeRedisOptions
/// HangfireDB /// HangfireDB
/// </summary> /// </summary>
public string? HangfireDB { get; set; } public string? HangfireDB { get; set; }
}
} }