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:
ChenYi 2025-04-16 17:36:46 +08:00
parent 57123d653c
commit 1a275ec9c3
13 changed files with 1491 additions and 714 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

176
src/JiShe.CollectBus.Application.Contracts/RedisDataCache/IRedisDataCacheService.cs Normal file
View File

@ -0,0 +1,176 @@
using JiShe.CollectBus.Common.Models;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Application.Contracts
{
/// <summary>
/// 数据缓存服务接口
/// </summary>
public interface IRedisDataCacheService
{
/// <summary>
/// 单个添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="data">待缓存数据</param>
/// <returns></returns>
Task InsertDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T data) where T : DeviceCacheBasicModel;
/// <summary>
/// 批量添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="items">待缓存数据集合</param>
/// <returns></returns>
Task BatchInsertDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
IEnumerable<T> items) where T : DeviceCacheBasicModel;
/// <summary>
/// 删除缓存信息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="data">已缓存数据</param>
/// <returns></returns>
Task RemoveCacheDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T data) where T : DeviceCacheBasicModel;
/// <summary>
/// 修改缓存信息,映射关系未发生改变
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="newData">待修改缓存数据</param>
/// <returns></returns>
Task ModifyDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T newData) where T : DeviceCacheBasicModel;
/// <summary>
/// 修改缓存信息,映射关系已改变
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="oldMemberId">旧的映射关系</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="newData">待修改缓存数据</param>
/// <returns></returns>
Task ModifyDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string oldMemberId,
string redisZSetScoresIndexCacheKey,
T newData) where T : DeviceCacheBasicModel;
///// <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>
//Task<BusCacheGlobalPagedResult<T>> GetPagedData<T>(
//string redisHashCacheKey,
//string redisZSetScoresIndexCacheKey,
//IEnumerable<int> focusIds,
//int pageSize = 10,
//decimal? lastScore = null,
//string lastMember = null,
//bool descending = true)
//where T : DeviceCacheBasicModel;
/// <summary>
/// 通过ZSET索引获取数据
/// </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>
Task<BusCacheGlobalPagedResult<T>> GetAllPagedData<T>(
string redisHashCacheKey,
string redisZSetScoresIndexCacheKey,
int pageSize = 1000,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel;
///// <summary>
///// 游标分页查询
///// </summary>
///// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
///// <param name="pageSize">分页数量</param>
///// <param name="startScore">开始索引</param>
///// <param name="excludeMember">开始唯一标识</param>
///// <param name="descending">排序方式</param>
///// <returns></returns>
//Task<(List<string> Members, bool HasNext)> GetPagedMembers(
// string redisZSetScoresIndexCacheKey,
// int pageSize,
// decimal? startScore,
// string excludeMember,
// bool descending);
///// <summary>
///// 批量获取指定分页的数据
///// </summary>
///// <typeparam name="T"></typeparam>
///// <param name="redisHashCacheKey">Hash表缓存key</param>
///// <param name="members">Hash表字段集合</param>
///// <returns></returns>
//Task<Dictionary<string, T>> BatchGetData<T>(
// string redisHashCacheKey,
// IEnumerable<string> members)
// where T : DeviceCacheBasicModel;
///// <summary>
///// 获取下一页游标
///// </summary>
///// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
///// <param name="lastMember">最后一个唯一标识</param>
///// <param name="descending">排序方式</param>
///// <returns></returns>
//Task<decimal?> GetNextScore(
// string redisZSetScoresIndexCacheKey,
// string lastMember,
// bool descending);
}
}

4
src/JiShe.CollectBus.Application/CollectBusAppService.cs
View File

@ -79,7 +79,7 @@ public abstract class CollectBusAppService : ApplicationService
{
string key = (string)item[0];
object[] fieldsAndValues = (object[])item[1];
var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, systemType, serverTagName, meterType, timeDensity)}";
var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, systemType, serverTagName, meterType, timeDensity)}";
string focusAddress = key.Replace(redisCacheKey, "");
var meterHashs = new Dictionary<string, T>();
@ -182,7 +182,7 @@ public abstract class CollectBusAppService : ApplicationService
string key = (string)item[0];
object[] fieldsAndValues = (object[])item[1];
var redisCacheKey = string.Format(
RedisConst.CacheMeterInfoKey,
RedisConst.CacheMeterInfoHashKey,
systemType,
serverTagName,
meterType,

745
src/JiShe.CollectBus.Application/RedisDataCache/RedisDataCacheService.cs Normal file
View File

@ -0,0 +1,745 @@
using Confluent.Kafka;
using FreeRedis;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.FreeRedisProvider;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Volo.Abp.DependencyInjection;
namespace JiShe.CollectBus.RedisDataCache
{
/// <summary>
/// 数据缓存服务接口
/// </summary>
public class RedisDataCacheService : IRedisDataCacheService, ITransientDependency
{
private readonly IFreeRedisProvider _freeRedisProvider;
private readonly ILogger<RedisDataCacheService> _logger;
private RedisClient Instance { get; set; }
/// <summary>
/// 数据缓存服务接口
/// </summary>
/// <param name="freeRedisProvider"></param>
/// <param name="logger"></param>
public RedisDataCacheService(IFreeRedisProvider freeRedisProvider,
ILogger<RedisDataCacheService> logger)
{
this._freeRedisProvider = freeRedisProvider;
this._logger = logger;
Instance = _freeRedisProvider.Instance;
}
/// <summary>
/// 单个添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="data">待缓存数据</param>
/// <returns></returns>
public async Task InsertDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T data) where T : DeviceCacheBasicModel
{
// 参数校验增强
if (data == null || string.IsNullOrWhiteSpace(redisHashCacheKey)
|| string.IsNullOrWhiteSpace(redisSetIndexCacheKey)
|| string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
{
_logger.LogError($"{nameof(InsertDataAsync)} 参数异常，-101");
return;
}
// 使用事务保证原子性
using (var trans = Instance.Multi())
{
// 主数据存储Hash
trans.HSet(redisHashCacheKey, data.MemberID, data.Serialize());
// 集中器号分组索引Set缓存
trans.SAdd(redisSetIndexCacheKey, data.MemberID);
// 集中器与表计信息排序索引ZSET缓存Key
trans.ZAdd(redisZSetScoresIndexCacheKey, data.ScoreValue, data.MemberID);
var results = trans.Exec();
if (results == null || results.Length <= 0)
{
_logger.LogError($"{nameof(InsertDataAsync)} 添加事务提交失败，-102");
}
}
await Task.CompletedTask;
}
/// <summary>
/// 批量添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="items">待缓存数据集合</param>
/// <returns></returns>
public async Task BatchInsertDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
IEnumerable<T> items) where T : DeviceCacheBasicModel
{
if (items == null
|| items.Count() <= 0
|| string.IsNullOrWhiteSpace(redisHashCacheKey)
|| string.IsNullOrWhiteSpace(redisSetIndexCacheKey)
|| string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
{
_logger.LogError($"{nameof(BatchInsertDataAsync)} 参数异常，-101");
return;
}
const int BATCH_SIZE = 1000; // 每批1000条
var semaphore = new SemaphoreSlim(Environment.ProcessorCount * 2);
foreach (var batch in items.Batch(BATCH_SIZE))
{
await semaphore.WaitAsync();
_ = Task.Run(() =>
{
using (var pipe = Instance.StartPipe())
{
foreach (var item in batch)
{
// 主数据存储Hash
pipe.HSet(redisHashCacheKey, item.MemberID, item.Serialize());
// Set索引缓存
pipe.SAdd(redisSetIndexCacheKey, item.MemberID);
// ZSET索引缓存Key
pipe.ZAdd(redisZSetScoresIndexCacheKey, item.ScoreValue, item.MemberID);
}
pipe.EndPipe();
}
semaphore.Release();
});
}
await Task.CompletedTask;
}
/// <summary>
/// 删除缓存信息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="data">已缓存数据</param>
/// <returns></returns>
public async Task RemoveCacheDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T data) where T : DeviceCacheBasicModel
{
if (data == null
|| string.IsNullOrWhiteSpace(redisHashCacheKey)
|| string.IsNullOrWhiteSpace(redisSetIndexCacheKey)
|| string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
{
_logger.LogError($"{nameof(RemoveCacheDataAsync)} 参数异常，-101");
return;
}
const string luaScript = @"
local hashCacheKey = KEYS[1]
local setIndexCacheKey = KEYS[2]
local zsetScoresIndexCacheKey = KEYS[3]
local member = ARGV[1]
local deleted = 0
if redis.call('HDEL', hashCacheKey, member) > 0 then
deleted = 1
end
redis.call('SREM', setIndexCacheKey, member)
redis.call('ZREM', zsetScoresIndexCacheKey, member)
return deleted
";
var keys = new[]
{
redisHashCacheKey,
redisSetIndexCacheKey,
redisZSetScoresIndexCacheKey
};
var result = await Instance.EvalAsync(luaScript, keys, new[] { data.MemberID });
if ((int)result == 0)
{
_logger.LogError($"{nameof(RemoveCacheDataAsync)} 删除指定Key{redisHashCacheKey}的{data.MemberID}数据失败,-102");
}
}
/// <summary>
/// 修改缓存信息,映射关系未发生改变
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="newData">待修改缓存数据</param>
/// <returns></returns>
public async Task ModifyDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
T newData) where T : DeviceCacheBasicModel
{
if (newData == null
|| string.IsNullOrWhiteSpace(redisHashCacheKey)
|| string.IsNullOrWhiteSpace(redisSetIndexCacheKey)
|| string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
{
_logger.LogError($"{nameof(ModifyDataAsync)} 参数异常，-101");
return;
}
var luaScript = @"
local hashCacheKey = KEYS[1]
local member = ARGV[1]
local newData = ARGV[2]
-- 校验存在性
if redis.call('HEXISTS', hashCacheKey, member) == 0 then
return 0
end
-- 更新主数据
redis.call('HSET', hashCacheKey, member, newData)
return 1
";
var result = await Instance.EvalAsync(luaScript,
new[]
{
redisHashCacheKey
},
new object[]
{
newData.MemberID,
newData.Serialize()
});
if ((int)result == 0)
{
_logger.LogError($"{nameof(ModifyDataAsync)} 更新指定Key{redisHashCacheKey}的{newData.MemberID}数据失败,-102");
}
}
/// <summary>
/// 修改缓存信息,映射关系已经改变
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
/// <param name="oldMemberId">旧的映射关系</param>
/// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
/// <param name="newData">待修改缓存数据</param>
/// <returns></returns>
public async Task ModifyDataAsync<T>(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string oldMemberId,
string redisZSetScoresIndexCacheKey,
T newData) where T : DeviceCacheBasicModel
{
if (newData == null
|| string.IsNullOrWhiteSpace(redisHashCacheKey)
|| string.IsNullOrWhiteSpace(redisSetIndexCacheKey)
|| string.IsNullOrWhiteSpace(oldMemberId)
|| string.IsNullOrWhiteSpace(redisZSetScoresIndexCacheKey))
{
_logger.LogError($"{nameof(ModifyDataAsync)} 参数异常，-101");
return;
}
var luaScript = @"
local hashCacheKey = KEYS[1]
local setIndexCacheKey = KEYS[2]
local zsetScoresIndexCacheKey = KEYS[3]
local member = ARGV[1]
local oldMember = ARGV[2]
local newData = ARGV[3]
local newScore = ARGV[4]
-- 校验存在性
if redis.call('HEXISTS', hashCacheKey, oldMember) == 0 then
return 0
end
-- 删除旧数据
redis.call('HDEL', hashCacheKey, oldMember)
-- 插入新主数据
redis.call('HSET', hashCacheKey, member, newData)
-- 处理变更
if newScore ~= '' then
-- 删除旧索引
redis.call('SREM', setIndexCacheKey, oldMember)
redis.call('ZREM', zsetScoresIndexCacheKey, oldMember)
-- 添加新索引
redis.call('SADD', setIndexCacheKey, member)
redis.call('ZADD', zsetScoresIndexCacheKey, newScore, member)
end
return 1
";
var result = await Instance.EvalAsync(luaScript,
new[]
{
redisHashCacheKey,
redisSetIndexCacheKey,
redisZSetScoresIndexCacheKey
},
new object[]
{
newData.MemberID,
oldMemberId,
newData.Serialize(),
newData.ScoreValue.ToString() ?? "",
});
if ((int)result == 0)
{
_logger.LogError($"{nameof(ModifyDataAsync)} 更新指定Key{redisHashCacheKey}的{newData.MemberID}数据失败,-102");
}
}
/// <summary>
/// 通过集中器与表计信息排序索引获取指定集中器号集合数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisCacheFocusScoresIndexKey">集中器与表计信息排序索引ZSET缓存Key</param>
/// <param name="focusIds">集中器Id</param>
/// <param name="pageSize">分页尺寸</param>
/// <param name="lastScore">最后一个索引</param>
/// <param name="lastMember">最后一个唯一标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
public async Task<BusCacheGlobalPagedResult<T>> GetPagedData<T>(
string redisCacheKey,
string redisCacheFocusScoresIndexKey,
IEnumerable<int> focusIds,
int pageSize = 10,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel
{
throw new Exception();
}
/// <summary>
/// 通过ZSET索引获取数据
/// </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 luaScript = @"
local command = ARGV[1]
local range_start = ARGV[2]
local range_end = ARGV[3]
local limit = tonumber(ARGV[4])
local last_score = ARGV[5]
local last_member = ARGV[6]
-- 处理相同分数下的字典序分页
if last_score ~= '' and last_member ~= '' then
if command == 'ZRANGEBYSCORE' then
range_start = '(' .. last_score
range_end = '(' .. last_score .. ' ' .. last_member
else
range_start = '(' .. last_score .. ' ' .. last_member
range_end = '(' .. last_score
end
end
-- 执行范围查询
local members
if command == 'ZRANGEBYSCORE' then
members = redis.call(command, KEYS[1], range_start, range_end,
'WITHSCORES', 'LIMIT', 0, limit)
else
members = redis.call(command, KEYS[1], range_end, range_start,
'WITHSCORES', 'LIMIT', 0, limit)
end
-- 提取成员和分数
local result_members = {}
local result_scores = {}
for i = 1, #members, 2 do
table.insert(result_members, members[i])
table.insert(result_scores, members[i+1])
end
-- 获取Hash数据
local hash_data = redis.call('HMGET', KEYS[2], unpack(result_members))
return {
#result_members,
result_members,
result_scores,
hash_data
}";
//正确设置范围参数
string rangeStart, rangeEnd;
if (descending)
{
rangeStart = lastScore.HasValue ? $"({lastScore}" : "+inf";
rangeEnd = "-inf"; // 降序时固定为最小值
}
else
{
rangeStart = lastScore.HasValue ? $"({lastScore}" : "-inf";
rangeEnd = "+inf"; // 升序时固定为最大值
}
var result = (object[])await Instance.EvalAsync(
luaScript,
new[] { redisZSetScoresIndexCacheKey, redisHashCacheKey },
new object[]
{
descending ? "ZREVRANGEBYSCORE" : "ZRANGEBYSCORE",
rangeStart,
rangeEnd,
(pageSize + 1).ToString(), // 多取1条用于判断hasNext
lastScore?.ToString() ?? "",
lastMember ?? ""
});
if ((long)result[0] == 0)
return new BusCacheGlobalPagedResult<T> { Items = new List<T>() };
// 处理结果集
var members = ((object[])result[1]).Cast<string>().ToList();
var scores = ((object[])result[2]).Cast<string>().Select(decimal.Parse).ToList();
var hashData = ((object[])result[3]).Cast<string>().ToList();
//合并有效数据并处理游标
var validItems = members.Zip(hashData, (m, h) =>
!string.IsNullOrWhiteSpace(h) ? BusJsonSerializer.Deserialize<T>(h) : null)
.Where(x => x != null)
.Take(pageSize + 1)
.ToList();
var hasNext = validItems.Count > pageSize;
var actualItems = hasNext ? validItems.Take(pageSize) : validItems;
// 计算下一页起始点
string nextMember = null;
decimal? nextScore = null;
if (hasNext)
{
// 获取实际返回的最后一条有效数据
var lastValidIndex = actualItems.Count() - 1;
nextMember = members[lastValidIndex];
nextScore = scores[lastValidIndex];
}
return new BusCacheGlobalPagedResult<T>
{
Items = actualItems.ToList(),
HasNext = hasNext,
NextScore = nextScore,
NextMember = nextMember,
TotalCount = await GetTotalCount(redisZSetScoresIndexCacheKey),
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>
/// 游标分页查询
/// </summary>
/// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
/// <param name="pageSize">分页数量</param>
/// <param name="lastScore">上一个索引</param>
/// <param name="lastMember">上一个标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
private async Task<(List<string> Members, bool HasNext, decimal? NextScore, string NextMember)> GetPagedMembers(
string redisZSetScoresIndexCacheKey,
int pageSize,
decimal? lastScore,
string lastMember,
bool descending)
{
// 根据排序方向初始化参数
long initialScore = descending ? long.MaxValue : 0;
decimal? currentScore = lastScore ?? initialScore;
string currentMember = lastMember;
var members = new List<string>(pageSize + 1);
// 使用游标分页查询
while (members.Count < pageSize + 1 && currentScore.HasValue)
{
var (batch, hasMore) = await GetNextBatch(
redisZSetScoresIndexCacheKey,
pageSize + 1 - members.Count,
currentScore.Value,
currentMember,
descending);
if (!batch.Any()) break;
members.AddRange(batch);
// 更新游标
currentMember = batch.LastOrDefault();
currentScore = await GetNextScore(redisZSetScoresIndexCacheKey, currentMember, descending);
}
// 处理分页结果
bool hasNext = members.Count > pageSize;
var resultMembers = members.Take(pageSize).ToList();
return (
resultMembers,
hasNext,
currentScore,
currentMember
);
}
/// <summary>
/// 批量获取指定分页的数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisHashCacheKey">Hash表缓存key</param>
/// <param name="members">Hash表字段集合</param>
/// <returns></returns>
private async Task<Dictionary<string, T>> BatchGetData<T>(
string redisHashCacheKey,
IEnumerable<string> members)
where T : DeviceCacheBasicModel
{
using var pipe = Instance.StartPipe();
foreach (var member in members)
{
pipe.HGet<T>(redisHashCacheKey, member);
}
var results = pipe.EndPipe();
return await Task.FromResult(members.Zip(results, (k, v) => new { k, v })
.ToDictionary(x => x.k, x => (T)x.v));
}
/// <summary>
/// 处理下一个分页数据
/// </summary>
/// <param name="zsetKey"></param>
/// <param name="limit"></param>
/// <param name="score"></param>
/// <param name="excludeMember"></param>
/// <param name="descending"></param>
/// <returns></returns>
private async Task<(string[] Batch, bool HasMore)> GetNextBatch(
string zsetKey,
int limit,
decimal score,
string excludeMember,
bool descending)
{
var query = descending
? await Instance.ZRevRangeByScoreAsync(
zsetKey,
max: score,
min: 0,
offset: 0,
count: limit)
: await Instance.ZRangeByScoreAsync(
zsetKey,
min: score,
max: long.MaxValue,
offset: 0,
count: limit);
return (query, query.Length >= limit);
}
/// <summary>
/// 获取下一页游标
/// </summary>
/// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
/// <param name="lastMember">最后一个唯一标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
private async Task<decimal?> GetNextScore(
string redisZSetScoresIndexCacheKey,
string lastMember,
bool descending)
{
if (string.IsNullOrEmpty(lastMember)) return null;
var score = await Instance.ZScoreAsync(redisZSetScoresIndexCacheKey, lastMember);
if (!score.HasValue) return null;
// 根据排序方向调整score
return descending
? score.Value - 1 // 降序时下页查询小于当前score
: score.Value + 1; // 升序时下页查询大于当前score
}
/// <summary>
/// 获取指定ZSET区间内的总数量
/// </summary>
/// <param name="zsetKey"></param>
/// <param name="min"></param>
/// <param name="max"></param>
/// <returns></returns>
public async Task<long> GetCount(string zsetKey, long min, long max)
{
// 缓存计数优化
var cacheKey = $"{zsetKey}_count_{min}_{max}";
var cached = await Instance.GetAsync<long?>(cacheKey);
if (cached.HasValue)
return cached.Value;
var count = await Instance.ZCountAsync(zsetKey, min, max);
await Instance.SetExAsync(cacheKey, 60, count); // 缓存60秒
return count;
}
/// <summary>
/// 获取指定ZSET的总数量
/// </summary>
/// <param name="redisZSetScoresIndexCacheKey"></param>
/// <returns></returns>
private async Task<long> GetTotalCount(string redisZSetScoresIndexCacheKey)
{
// 缓存计数优化
var cacheKey = $"{redisZSetScoresIndexCacheKey}_total_count";
var cached = await Instance.GetAsync<long?>(cacheKey);
if (cached.HasValue)
return cached.Value;
var count = await Instance.ZCountAsync(redisZSetScoresIndexCacheKey, 0, decimal.MaxValue);
await Instance.SetExAsync(cacheKey, 30, count); // 缓存30秒
return count;
}
}
}

57
src/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -23,6 +23,9 @@ using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Kafka.Attributes;
using System.Text.Json;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Models;
using System.Diagnostics;
namespace JiShe.CollectBus.Samples;
@ -32,17 +35,23 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
private readonly IIoTDBProvider _iotDBProvider;
private readonly IoTDBRuntimeContext _dbContext;
private readonly IoTDBOptions _options;
private readonly IRedisDataCacheService _redisDataCacheService;
public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options,
IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger)
IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService)
{
_iotDBProvider = iotDBProvider;
_options = options.Value;
_dbContext = dbContext;
_logger = logger;
_redisDataCacheService = redisDataCacheService;
}
/// <summary>
/// 测试 UseSessionPool
/// </summary>
/// <param name="timestamps"></param>
/// <returns></returns>
[HttpGet]
public async Task UseSessionPool(long timestamps)
{
@ -72,7 +81,10 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
await _iotDBProvider.InsertAsync(meter);
}
/// <summary>
/// 测试Session切换
/// </summary>
/// <returns></returns>
[HttpGet]
public async Task UseTableSessionPool()
{
@ -125,7 +137,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
var timeDensity = "15";
//获取缓存中的电表信息
var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";
var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";
var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, "Energy", "JiSheCollectBus", timeDensity, MeterTypeEnum.Ammeter);
@ -178,6 +190,43 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
await _iotDBProvider.InsertAsync(meter);
}
/// <summary>
/// 测试单个测点数据项
/// </summary>
/// <returns></returns>
[HttpGet]
public async Task TestRedisCacheGetAllPagedData()
{
var timeDensity = "15";
string SystemType = "";
string ServerTagName = "JiSheCollectBus2";
var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
var timer = Stopwatch.StartNew();
decimal? cursor = null;
string member = null;
bool hasNext;
List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
do
{
var page = await _redisDataCacheService
.GetAllPagedData<AmmeterInfo>(
redisCacheMeterInfoHashKeyTemp,
redisCacheMeterInfoZSetScoresIndexKeyTemp);
meterInfos.AddRange(page.Items);
cursor = page.NextScore;
member = page.NextMember;
hasNext = page.HasNext;
} while (hasNext);
timer.Stop();
_logger.LogInformation($"{nameof(TestRedisCacheGetAllPagedData)} 获取电表缓存数据完成，耗时{timer.ElapsedMilliseconds}毫秒");
}
public Task<SampleDto> GetAsync()
{

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

@ -1,11 +1,13 @@
using DotNetCore.CAP;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.BuildSendDatas;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.MessageIssueds;
@ -34,13 +36,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading
private readonly IIoTDBProvider _dbProvider;
private readonly IMeterReadingRecordRepository _meterReadingRecordRepository;
private readonly IProducerService _producerService;
private readonly IRedisDataCacheService _redisDataCacheService;
public BasicScheduledMeterReadingService(
ILogger<BasicScheduledMeterReadingService> logger,
ICapPublisher producerBus,
IMeterReadingRecordRepository meterReadingRecordRepository,
IProducerService producerService,
IRedisDataCacheService redisDataCacheService,
IIoTDBProvider dbProvider)
{
_producerBus = producerBus;
@ -48,6 +51,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_dbProvider = dbProvider;
_meterReadingRecordRepository = meterReadingRecordRepository;
_producerService = producerService;
_redisDataCacheService = redisDataCacheService;
}
/// <summary>
@ -121,7 +125,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//获取缓存中的表信息
var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, meteryType, timeDensity)}*";
var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, meteryType, timeDensity)}*";
var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
{
@ -209,25 +213,51 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public virtual async Task InitAmmeterCacheData(string gatherCode = "")
{
#if DEBUG
var timeDensity = "15";
string tempCacheMeterInfoKey = $"CollectBus:{"{0}:{1}"}:MeterInfo:{"{2}"}:{"{3}"}";
//获取缓存中的电表信息
var redisKeyList = $"{string.Format(tempCacheMeterInfoKey, SystemType, "JiSheCollectBus", MeterTypeEnum.Ammeter, timeDensity)}*";
//var timeDensity = "15";
//string tempCacheMeterInfoKey = $"CollectBus:{"{0}:{1}"}:MeterInfo:{"{2}"}:{"{3}"}";
////获取缓存中的电表信息
//var redisKeyList = $"{string.Format(tempCacheMeterInfoKey, SystemType, "JiSheCollectBus", MeterTypeEnum.Ammeter, timeDensity)}*";
var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
var tempMeterInfos = await GetMeterRedisCacheListData<AmmeterInfoTemp>(oneMinutekeyList, SystemType, ServerTagName, timeDensity, MeterTypeEnum.Ammeter);
//List<string> focusAddressDataLista = new List<string>();
List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
foreach (var item in tempMeterInfos)
{
var tempData = item.Adapt<AmmeterInfo>();
tempData.FocusId = item.FocusID;
tempData.MeterId = item.Id;
meterInfos.Add(tempData);
//focusAddressDataLista.Add(item.FocusAddress);
}
//var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
//var tempMeterInfos = await GetMeterRedisCacheListData<AmmeterInfoTemp>(oneMinutekeyList, SystemType, ServerTagName, timeDensity, MeterTypeEnum.Ammeter);
////List<string> focusAddressDataLista = new List<string>();
//List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
//foreach (var item in tempMeterInfos)
//{
// var tempData = item.Adapt<AmmeterInfo>();
// tempData.FocusId = item.FocusID;
// tempData.MeterId = item.Id;
// meterInfos.Add(tempData);
// //focusAddressDataLista.Add(item.FocusAddress);
//}
//DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);
var timeDensity = "15";
var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
decimal? cursor = null;
string member = null;
bool hasNext;
List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
do
{
var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
redisCacheMeterInfoHashKeyTemp,
redisCacheMeterInfoZSetScoresIndexKeyTemp,
pageSize: 1000,
lastScore: cursor,
lastMember: member);
meterInfos.AddRange(page.Items);
cursor = page.HasNext ? page.NextScore : null;
member = page.HasNext ? page.NextMember : null;
hasNext = page.HasNext;
} while (hasNext);
#else
var meterInfos = await GetAmmeterInfoList(gatherCode);
#endif
@ -251,10 +281,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
{
var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheFocusIndexKey = $"{string.Format(RedisConst.CacheMeterInfoFocusIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheScoresIndexKey = $"{string.Format(RedisConst.CacheMeterInfoScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheGlobalIndexKey = $"{string.Format(RedisConst.CacheMeterInfoGlobalIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheMeterInfoHashKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheMeterInfoSetIndexKey = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
var redisCacheMeterInfoZSetScoresIndexKey = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
List<AmmeterInfo> ammeterInfos = new List<AmmeterInfo>();
//将表计信息根据集中器分组，获得集中器号
@ -329,11 +358,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
}
await FreeRedisProvider.BatchAddMeterData<AmmeterInfo>(
redisCacheKey,
redisCacheFocusIndexKey,
redisCacheScoresIndexKey,
redisCacheGlobalIndexKey, ammeterInfos);
await _redisDataCacheService.BatchInsertDataAsync<AmmeterInfo>(
redisCacheMeterInfoHashKey,
redisCacheMeterInfoSetIndexKey,
redisCacheMeterInfoZSetScoresIndexKey,ammeterInfos);
//在缓存表信息数据的时候，新增下一个时间的自动处理任务，1分钟后执行所有的采集频率任务
TasksToBeIssueModel nextTask = new TasksToBeIssueModel()
@ -635,7 +663,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
var currentTime = DateTime.Now;
var pendingCopyReadTime = currentTime.AddMinutes(timeDensity);
//构建缓存任务key，依然 表计类型+采集频率+集中器地址，存hash类型
var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{ammeterInfo.FocusAddress}";
var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{ammeterInfo.FocusAddress}";
if (string.IsNullOrWhiteSpace(ammeterInfo.ItemCodes))
{
@ -897,7 +925,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
foreach (var focusInfo in focusGroup)
{
//构建缓存任务key，依然 表计类型+采集频率+集中器地址，存hash类型
var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{focusInfo.Key}";
var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{focusInfo.Key}";
foreach (var ammeterInfo in focusInfo.Value)
{
@ -1113,7 +1141,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
continue;
}
var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.WaterMeter, itemTimeDensity.Key)}{item.Key}";
var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.WaterMeter, itemTimeDensity.Key)}{item.Key}";
Dictionary<string, WatermeterInfo> keyValuePairs = new Dictionary<string, WatermeterInfo>();
foreach (var subItem in item)
{
@ -1260,7 +1288,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
/// <returns></returns>
private string GetTelemetryPacketCacheKeyPrefix(int timeDensity, MeterTypeEnum meterType)
{
return $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, meterType, timeDensity)}*";
return $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, meterType, timeDensity)}*";
}
#endregion

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

@ -4,6 +4,7 @@ using System.Threading.Tasks;
using Confluent.Kafka;
using DotNetCore.CAP;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Common.Helpers;
@ -35,8 +36,19 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
{
string serverTagName = string.Empty;
public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger,
ICapPublisher producerBus, IIoTDBProvider dbProvider, IMeterReadingRecordRepository meterReadingRecordRepository,IConfiguration configuration, IProducerService producerService) : base(logger, producerBus, meterReadingRecordRepository, producerService,dbProvider)
public EnergySystemScheduledMeterReadingService(
ILogger<EnergySystemScheduledMeterReadingService> logger,
ICapPublisher producerBus, IIoTDBProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordRepository,
IConfiguration configuration,
IProducerService producerService,
IRedisDataCacheService redisDataCacheService)
: base(logger,
producerBus,
meterReadingRecordRepository,
producerService,
redisDataCacheService,
dbProvider)
{
serverTagName = configuration.GetValue<string>(CommonConst.ServerTagName)!;
}

29
src/JiShe.CollectBus.Common/Consts/RedisConst.cs
View File

@ -32,23 +32,18 @@ namespace JiShe.CollectBus.Common.Consts
/// <summary>
/// 缓存表计信息,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheMeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:{"{3}"}";
public const string CacheMeterInfoHashKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:{"{3}"}";
/// <summary>
/// 缓存表计信息集中器索引Set缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// 缓存表计信息索引Set缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheMeterInfoFocusIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:FocusIndex:{"{3}"}";
public const string CacheMeterInfoSetIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:SetIndex:{"{3}"}";
/// <summary>
/// 缓存表计信息集中器排序索引ZSET缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// 缓存表计信息排序索引ZSET缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheMeterInfoScoresIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:ScoresIndex:{"{3}"}";
/// <summary>
/// 缓存表计信息集中器采集频率分组全局索引ZSet缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheMeterInfoGlobalIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:GlobalIndex:{"{3}"}";
public const string CacheMeterInfoZSetScoresIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:ZSetScoresIndex:{"{3}"}";
public const string TaskInfo = "TaskInfo";
/// <summary>
@ -60,7 +55,17 @@ namespace JiShe.CollectBus.Common.Consts
/// <summary>
/// 缓存表计下发指令数据集,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheTelemetryPacketInfoKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:{"{3}"}";
public const string CacheTelemetryPacketInfoHashKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:{"{3}"}";
/// <summary>
/// 缓存表计下发指令数据集索引Set缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheTelemetryPacketInfoSetIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:SetIndex:{"{3}"}";
/// <summary>
/// 缓存表计下发指令数据集排序索引ZSET缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
/// </summary>
public const string CacheTelemetryPacketInfoZSetScoresIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:ZSetScoresIndex:{"{3}"}";
///// <summary>
///// 缓存设备平衡关系映射结果,{0}=>系统类型，{1}=>应用服务部署标记

16
src/JiShe.CollectBus.Common/Models/BusCacheGlobalPagedResult.cs
View File

@ -22,6 +22,22 @@ namespace JiShe.CollectBus.Common.Models
/// </summary>
public long TotalCount { get; set; }
/// <summary>
/// 每页条数
/// </summary>
public int PageSize { get; set; }
/// <summary>
/// 总页数
/// </summary>
public int PageCount
{
get
{
return (int)Math.Ceiling((double)TotalCount / PageSize);
}
}
/// <summary>
/// 是否有下一页
/// </summary>

7
src/JiShe.CollectBus.Common/Models/DeviceCacheBasicModel.cs
View File

@ -22,8 +22,13 @@ namespace JiShe.CollectBus.Common.Models
public int MeterId { get; set; }
/// <summary>
/// 唯一标识，是redis ZSet和Set memberid
/// 关系映射标识，用于ZSet的Member字段和Set的Value字段，具体值可以根据不同业务场景进行定义
/// </summary>
public virtual string MemberID => $"{FocusId}:{MeterId}";
/// <summary>
/// ZSet排序索引分数值，具体值可以根据不同业务场景进行定义，例如时间戳
/// </summary>
public virtual long ScoreValue=> ((long)FocusId << 32) | (uint)MeterId;
}
}

957
src/JiShe.CollectBus.FreeRedisProvider/FreeRedisProvider.cs
View File

File diff suppressed because it is too large Load Diff

100
src/JiShe.CollectBus.FreeRedisProvider/IFreeRedisProvider.cs
View File

@ -9,105 +9,7 @@ namespace JiShe.CollectBus.FreeRedisProvider
/// 获取客户端
/// </summary>
/// <returns></returns>
RedisClient Instance { get; set; }
/// <summary>
/// 单个添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
/// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
/// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
/// <param name="data">表计信息</param>
/// <param name="timestamp">可选时间戳</param>
/// <returns></returns>
Task AddMeterCacheData<T>(
string redisCacheKey,
string redisCacheFocusIndexKey,
string redisCacheScoresIndexKey,
string redisCacheGlobalIndexKey,
T data,
DateTimeOffset? timestamp = null) where T : DeviceCacheBasicModel;
/// <summary>
/// 批量添加数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
/// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
/// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
/// <param name="items">数据集合</param>
/// <param name="timestamp">可选时间戳</param>
/// <returns></returns>
Task BatchAddMeterData<T>(
string redisCacheKey,
string redisCacheFocusIndexKey,
string redisCacheScoresIndexKey,
string redisCacheGlobalIndexKey,
IEnumerable<T> items,
DateTimeOffset? timestamp = null) where T : DeviceCacheBasicModel;
/// <summary>
/// 删除指定redis缓存key的缓存数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
/// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
/// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
/// <param name="data">表计信息</param>
/// <returns></returns>
Task RemoveMeterData<T>(
string redisCacheKey,
string redisCacheFocusIndexKey,
string redisCacheScoresIndexKey,
string redisCacheGlobalIndexKey,
T data) where T : DeviceCacheBasicModel;
/// <summary>
/// 修改表计缓存信息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="oldRedisCacheFocusIndexKey">旧集中器索引Set缓存Key</param>
/// <param name="newRedisCacheFocusIndexKey">新集中器索引Set缓存Key</param>
/// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
/// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
/// <param name="newData">表计信息</param>
/// <param name="newTimestamp">可选时间戳</param>
/// <returns></returns>
Task UpdateMeterData<T>(
string redisCacheKey,
string oldRedisCacheFocusIndexKey,
string newRedisCacheFocusIndexKey,
string redisCacheScoresIndexKey,
string redisCacheGlobalIndexKey,
T newData,
DateTimeOffset? newTimestamp = null) where T : DeviceCacheBasicModel;
/// <summary>
/// 通过全局索引分页查询表计缓存数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
/// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
/// <param name="pageSize">分页尺寸</param>
/// <param name="lastScore">最后一个索引</param>
/// <param name="lastMember">最后一个唯一标识</param>
/// <param name="descending">排序方式</param>
/// <returns></returns>
Task<BusCacheGlobalPagedResult<T>> GetGlobalPagedData<T>(
string redisCacheKey,
string redisCacheGlobalIndexKey,
int pageSize = 10,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel;
RedisClient Instance { get; set; }
}
}

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

@ -13,6 +13,11 @@ namespace JiShe.CollectBus.FreeRedisProvider.Options
/// </summary>
public string? Configuration { get; set; }
/// <summary>
/// 最大连接数
/// </summary>
public string? MaxPoolSize { get; set; }
/// <summary>
/// 默认数据库
/// </summary>

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

@ -41,6 +41,7 @@
},
"Redis": {
"Configuration": "192.168.1.9:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"MaxPoolSize": "50",
"DefaultDB": "14",
"HangfireDB": "15"
},
@ -128,7 +129,7 @@
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
},
"ServerTagName": "JiSheCollectBus2",
"ServerTagName": "JiSheCollectBus3",
"KafkaReplicationFactor": 3,
"NumPartitions": 30
}