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:
陈益 2025-05-11 21:53:55 +08:00
parent f6130e1d0b
commit cf9bf6c210
6 changed files with 149 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/JiShe.CollectBus.IoTDB/Provider/IoTDBProvider.cs
View File

@ -730,7 +730,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
object rawValue = item1Member.Getter(obj);
string value = rawValue?.ToString();
if (!string.IsNullOrEmpty(value))
if (!string.IsNullOrWhiteSpace(value))
{
// 规则1: 严格检查ASCII字母和数字（0-9, A-Z, a-z）
bool hasInvalidChars = value.Any(c =>

2
services/JiShe.CollectBus.Application.Contracts/RedisDataCache/IRedisDataCacheService.cs
View File

@ -123,7 +123,7 @@ namespace JiShe.CollectBus.Application.Contracts
/// <param name="lastMember">最后一个唯一标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
Task<BusCacheGlobalPagedResult<T>> GetPagedData<T>(
Task<BusCacheGlobalPagedResult<T>> GetSingleData<T>(
string redisHashCacheKey,
string redisZSetScoresIndexCacheKey,
string scoreValueRawData,

175
services/JiShe.CollectBus.Application/RedisDataCache/RedisDataCacheService.cs
View File

@ -42,8 +42,6 @@ namespace JiShe.CollectBus.RedisDataCache
Instance = _freeRedisProvider.Instance;
}
//todo 单个数据查询
/// <summary>
/// 单个添加数据
/// </summary>
@ -370,7 +368,7 @@ namespace JiShe.CollectBus.RedisDataCache
/// <summary>
/// 通过集中器与表计信息排序索引获取数据
/// 通过集中器与表计信息排序索引获取单个数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
@ -381,24 +379,105 @@ namespace JiShe.CollectBus.RedisDataCache
/// <param name="lastMember">最后一个唯一标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
public async Task<BusCacheGlobalPagedResult<T>> GetPagedData<T>(
string redisHashCacheKey,
string redisZSetScoresIndexCacheKey,
string scoreValueRawData,
int pageSize = 10,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel
public async Task<BusCacheGlobalPagedResult<T>> GetSingleData<T>(
string redisHashCacheKey,
string redisZSetScoresIndexCacheKey,
string scoreValueRawData,
int pageSize = 10,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel
{
var rawDataArray = scoreValueRawData.Split(":");
string focusAddress = rawDataArray[0];
string point = rawDataArray[1];
// 参数校验
if (string.IsNullOrWhiteSpace(redisHashCacheKey) ||
string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey) ||
string.IsNullOrWhiteSpace(scoreValueRawData))
{
_logger.LogError($"{nameof(GetSingleData)} 参数异常，-101");
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
}
long scoreValue = 0;
// 解析原始数据
var rawDataArray = scoreValueRawData.Split(':');
if (rawDataArray.Length != 2)
{
_logger.LogError($"{nameof(GetSingleData)} scoreValueRawData格式错误，应为[focusAddress:point]");
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
}
// 计算Score值
long scoreValue;
try
{
var focusAddress = rawDataArray[0];
var point = Convert.ToInt32(rawDataArray[1]);
scoreValue = CommonHelper.GetFocusScores(focusAddress, point);
}
catch (Exception ex)
{
_logger.LogError(ex, $"{nameof(GetSingleData)} 计算Score值失败");
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
}
throw new Exception();
// Lua脚本：原子化查询操作
const string luaScript = @"
local zsetKey = KEYS[1]
local hashKey = KEYS[2]
local targetScore = ARGV[1]
-- 精确匹配Score并获取第一个成员
local members = redis.call('ZRANGEBYSCORE', zsetKey, targetScore, targetScore, 'LIMIT', 0, 1)
if #members == 0 then
return nil
end
-- 获取哈希表数据
local member = members[1]
local data = redis.call('HGET', hashKey, member)
return {member, data}";
try
{
// 执行脚本
var result = await Instance.EvalAsync(
luaScript,
new[] { redisZSetScoresIndexCacheKey, redisHashCacheKey },
new object[] { scoreValue });
// 处理空结果
if (result == null)
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
// 解析Redis返回数据
var resultArray = (object[])result;
var memberId = (string)resultArray[0];
var dataStr = (string)resultArray[1];
// 反序列化数据
var data = BusJsonSerializer.Deserialize<T>(dataStr);
if (data == null)
{
_logger.LogError($"{nameof(GetSingleData)} 反序列化失败，MemberId: {memberId}");
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
}
// 构造返回结果
return new BusCacheGlobalPagedResult<T>
{
Items = new List<T> { data },
TotalCount = 1,
PageSize = 1,
HasNext = false,
NextScore = null,
NextMember = null
};
}
catch (Exception ex)
{
_logger.LogError(ex, $"{nameof(GetSingleData)} Redis操作异常");
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
}
}
/// <summary>
@ -559,67 +638,7 @@ namespace JiShe.CollectBus.RedisDataCache
PageSize = pageSize,
};
}
///// <summary>
///// 通过集中器与表计信息排序索引获取数据
///// </summary>
///// <typeparam name="T"></typeparam>
///// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
///// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
///// <param name="pageSize">分页尺寸</param>
///// <param name="lastScore">最后一个索引</param>
///// <param name="lastMember">最后一个唯一标识</param>
///// <param name="descending">排序方式</param>
///// <returns></returns>
//public async Task<BusCacheGlobalPagedResult<T>> GetAllPagedData<T>(
//string redisHashCacheKey,
//string redisZSetScoresIndexCacheKey,
//int pageSize = 1000,
//decimal? lastScore = null,
//string lastMember = null,
//bool descending = true)
//where T : DeviceCacheBasicModel
//{
// // 参数校验增强
// if (string.IsNullOrWhiteSpace(redisHashCacheKey) || string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
// {
// _logger.LogError($"{nameof(GetAllPagedData)} 参数异常，-101");
// return null;
// }
// if (pageSize < 1 || pageSize > 10000)
// {
// _logger.LogError($"{nameof(GetAllPagedData)} 分页大小应在1-10000之间，-102");
// return null;
// }
// //// 分页参数解析
// //var (startScore, excludeMember) = descending
// // ? (lastScore ?? decimal.MaxValue, lastMember)
// // : (lastScore ?? 0, lastMember);
// //执行分页查询（整合游标处理）
// var pageResult = await GetPagedMembers(
// redisZSetScoresIndexCacheKey,
// pageSize,
// lastScore,
// lastMember,
// descending);
// // 批量获取数据（优化内存分配）
// var dataDict = await BatchGetData<T>(redisHashCacheKey, pageResult.Members);
// return new BusCacheGlobalPagedResult<T>
// {
// Items = pageResult.Members.Select(m => dataDict.TryGetValue(m, out var v) ? v : default)
// .Where(x => x != null).ToList(),
// HasNext = pageResult.HasNext,
// NextScore = pageResult.NextScore,
// NextMember = pageResult.NextMember,
// TotalCount = await GetTotalCount(redisZSetScoresIndexCacheKey),
// PageSize = pageSize,
// };
//}
/// <summary>
/// 游标分页查询

2
services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -737,7 +737,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
ItemCode = tempItem,
DataTimeMark = new Protocol.DataTimeMark()
{
Density = ammeterInfo.TimeDensity.GetDensity(),//转换成协议的值
Density = ammeterInfo.TimeDensity.GetFocusDensity(),//转换成协议的值
Point = 1,
DataTime = timestamps,
}

2
services/JiShe.CollectBus.Domain/IotSystems/Ammeters/AmmeterInfo.cs
View File

@ -20,7 +20,7 @@ namespace JiShe.CollectBus.IotSystems.Ammeters
/// ZSet排序索引分数值，具体值可以根据不同业务场景进行定义，例如时间戳
/// </summary>
[Column(IsIgnore = true)]
public override long ScoreValue => ((long)FocusId << 32) | (uint)DateTime.Now.Ticks;
public override long ScoreValue => Common.Helpers.CommonHelper.GetFocusScores(FocusAddress,MeteringCode);
/// <summary>
/// 电表名称

61
shared/JiShe.CollectBus.Common/Helpers/CommonHelper.cs
View File

@ -10,6 +10,8 @@ using System.Text;
using System.Threading.Tasks;
using JiShe.CollectBus.Common.Attributes;
using System.Collections.Specialized;
using JiShe.CollectBus.Common.Enums;
using Newtonsoft.Json.Linq;
namespace JiShe.CollectBus.Common.Helpers
{
@ -796,7 +798,7 @@ namespace JiShe.CollectBus.Common.Helpers
/// <param name="msa"></param>
/// <param name="seq"></param>
/// <returns></returns>
public static string GetTaskMark(int afn, int fn, int pn, int msa,int seq)
public static string GetTaskMark(int afn, int fn, int pn, int msa, int seq)
{
var makstr = $"{afn.ToString().PadLeft(2, '0')}{fn.ToString().PadLeft(2, '0')}{pn.ToString().PadLeft(2, '0')}{msa.ToString().PadLeft(2, '0')}{seq.ToString().PadLeft(2, '0')}";
@ -860,20 +862,53 @@ namespace JiShe.CollectBus.Common.Helpers
}
/// <summary>
/// 采集频率转换为集中器采集密度
/// 系统采集频率转换为集中器采集密度
/// </summary>
/// <param name="timeDensity"></param>
/// <returns></returns>
public static int GetDensity(this int timeDensity) =>
timeDensity switch
{
0 => 0,//无
1 => 255,//1分钟
5 => 245,//5分钟
15 => 1,//15分钟
30 => 2,//30分钟
60 => 3,//60分钟
_ => -1//采集项本身无密度位
};
public static int GetFocusDensity(this int timeDensity) => timeDensity switch
{
0 => 0,//无
1 => 255,//1分钟
5 => 245,//5分钟
15 => 1,//15分钟
30 => 2,//30分钟
60 => 3,//60分钟
_ => -1//采集项本身无密度位
};
/// <summary>
/// 集中器采集密度转换为系统采集频率
/// </summary>
/// <param name="density"></param>
/// <returns></returns>
public static int GetSystemDensity(this int density) => density switch
{
0 => 0,//无
255 => 1,//1分钟
245 => 5,//5分钟
1 => 15,//15分钟
2 => 30,//30分钟
3 => 60,//60分钟
_ => -1//采集项本身无密度位
};
/// <summary>
/// 获取集中器ZSet Scores
/// </summary>
/// <param name="focusScores"></param>
/// <param name="point"></param>
/// <returns></returns>
public static long GetFocusScores(string focusScores, int point)
{
bool hasInvalidChars = focusScores.Any(c => !(c >= '0' && c <= '9'));
if (hasInvalidChars)
{
throw new Exception($"{nameof(GetFocusScores)} 集中器地址格式错误");
}
var scoresStr = $"{focusScores}{point.ToString().PadLeft(2, '0')}";
return Convert.ToInt64(scoresStr);
}
}
}