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 Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using JiShe.CollectBus.FreeRedis;
using Volo.Abp.DependencyInjection;
using static FreeSql.Internal.GlobalFilter;
using static System.Runtime.InteropServices.JavaScript.JSType;
using static Volo.Abp.UI.Navigation.DefaultMenuNames.Application;
namespace JiShe.CollectBus.RedisDataCache
{
///
/// 数据缓存服务接口
///
public class RedisDataCacheService : IRedisDataCacheService, ITransientDependency
{
private readonly IFreeRedisProvider _freeRedisProvider;
private readonly ILogger _logger;
private RedisClient Instance { get; set; }
///
/// 数据缓存服务接口
///
///
///
public RedisDataCacheService(IFreeRedisProvider freeRedisProvider,
ILogger logger)
{
this._freeRedisProvider = freeRedisProvider;
this._logger = logger;
Instance = _freeRedisProvider.Instance;
}
///
/// 单个添加数据
///
///
/// 主数据存储Hash缓存Key
/// Set索引缓存Key
/// ZSET索引缓存Key
/// 待缓存数据
///
public async Task InsertDataAsync(
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;
}
///
/// 批量添加数据
///
///
/// 主数据存储Hash缓存Key
/// Set索引缓存Key
/// ZSET索引缓存Key
/// 待缓存数据集合
///
public async Task BatchInsertDataAsync(
string redisHashCacheKey,
string redisSetIndexCacheKey,
string redisZSetScoresIndexCacheKey,
IEnumerable 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;
}
///
/// 删除缓存信息
///
///
/// 主数据存储Hash缓存Key
/// Set索引缓存Key
/// ZSET索引缓存Key
/// 已缓存数据
///
public async Task RemoveCacheDataAsync(
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");
}
}
///
/// 修改缓存信息,映射关系未发生改变
///
///
/// 主数据存储Hash缓存Key
/// Set索引缓存Key
/// ZSET索引缓存Key
/// 待修改缓存数据
///
public async Task ModifyDataAsync(
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");
}
}
///
/// 修改缓存信息,映射关系已经改变
///
///
/// 主数据存储Hash缓存Key
/// Set索引缓存Key
/// 旧的映射关系
/// ZSET索引缓存Key
/// 待修改缓存数据
///
public async Task ModifyDataAsync(
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");
}
}
///
/// 通过集中器与表计信息排序索引获取指定集中器号集合数据
///
///
/// 主数据存储Hash缓存Key
/// 集中器与表计信息排序索引ZSET缓存Key
/// 集中器Id
/// 分页尺寸
/// 最后一个索引
/// 最后一个唯一标识
/// 排序方式
///
public async Task> GetPagedData(
string redisCacheKey,
string redisCacheFocusScoresIndexKey,
IEnumerable focusIds,
int pageSize = 10,
decimal? lastScore = null,
string lastMember = null,
bool descending = true)
where T : DeviceCacheBasicModel
{
throw new Exception();
}
///
/// 通过ZSET索引获取数据,支持10万级别数据处理,控制在13秒以内。
///
///
/// 主数据存储Hash缓存Key
/// ZSET索引缓存Key
/// 分页尺寸
/// 最后一个索引
/// 最后一个唯一标识
/// 排序方式
///
public async Task> GetAllPagedData(
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 new BusCacheGlobalPagedResult { Items = new List() };
}
pageSize = Math.Clamp(pageSize, 1, 10000);
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]
-- 获取原始数据
local members
if command == 'ZRANGEBYSCORE' then
members = redis.call(command, KEYS[1], range_start, range_end, 'WITHSCORES', 'LIMIT', 0, limit * 2)
else
members = redis.call('ZREVRANGEBYSCORE', KEYS[1], range_start, range_end, 'WITHSCORES', 'LIMIT', 0, limit * 2)
end
-- 过滤数据
local filtered_members = {}
local count = 0
for i = 1, #members, 2 do
local member = members[i]
local score = members[i+1]
local include = true
if last_score ~= '' and last_member ~= '' then
if command == 'ZRANGEBYSCORE' then
-- 升序:score > last_score 或 (score == last_score 且 member > last_member)
if score == last_score then
include = member > last_member
else
include = tonumber(score) > tonumber(last_score)
end
else
-- 降序:score < last_score 或 (score == last_score 且 member < last_member)
if score == last_score then
include = member < last_member
else
include = tonumber(score) < tonumber(last_score)
end
end
end
if include then
table.insert(filtered_members, member)
table.insert(filtered_members, score)
count = count + 1
if count >= limit then
break
end
end
end
-- 提取有效数据
local result_members, result_scores = {}, {}
for i=1,#filtered_members,2 do
table.insert(result_members, filtered_members[i])
table.insert(result_scores, filtered_members[i+1])
end
if #result_members == 0 then return {0,{},{},{}} 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.Value.ToString() : "+inf";
rangeEnd = "-inf";
}
else
{
rangeStart = lastScore.HasValue ? lastScore.Value.ToString() : "-inf";
rangeEnd = "+inf";
}
var scriptResult = (object[])await Instance.EvalAsync(luaScript,
new[] { redisZSetScoresIndexCacheKey, redisHashCacheKey },
new object[]
{
descending ? "ZREVRANGEBYSCORE" : "ZRANGEBYSCORE",
rangeStart,
rangeEnd,
(pageSize + 1).ToString(), // 获取pageSize+1条以判断是否有下一页
lastScore?.ToString() ?? "",
lastMember ?? ""
});
if ((long)scriptResult[0] == 0)
return new BusCacheGlobalPagedResult { Items = new List() };
// 处理结果集
var members = ((object[])scriptResult[1]).Cast().ToList();
var scores = ((object[])scriptResult[2]).Cast().Select(decimal.Parse).ToList();
var hashData = ((object[])scriptResult[3]).Cast().ToList();
var validItems = members.AsParallel()
.Select((m, i) =>
{
try { return BusJsonSerializer.Deserialize(hashData[i]); }
catch { return null; }
})
.Where(x => x != null)
.ToList();
var hasNext = validItems.Count > pageSize;
var actualItems = hasNext ? validItems.Take(pageSize) : validItems;
//分页锚点索引
decimal? nextScore = null;
string nextMember = null;
if (hasNext && actualItems.Any())
{
var lastIndex = actualItems.Count() - 1; // 使用actualItems的最后一个索引
nextScore = scores[lastIndex];
nextMember = members[lastIndex];
}
return new BusCacheGlobalPagedResult
{
Items = actualItems.ToList(),
HasNext = hasNext,
NextScore = nextScore,
NextMember = nextMember,
TotalCount = await GetTotalCount(redisZSetScoresIndexCacheKey),
PageSize = pageSize,
};
}
/////
///// 通过集中器与表计信息排序索引获取数据
/////
/////
///// 主数据存储Hash缓存Key
///// ZSET索引缓存Key
///// 分页尺寸
///// 最后一个索引
///// 最后一个唯一标识
///// 排序方式
/////
//public async Task> GetAllPagedData(
//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(redisHashCacheKey, pageResult.Members);
// return new BusCacheGlobalPagedResult
// {
// 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,
// };
//}
///
/// 游标分页查询
///
/// 排序索引ZSET缓存Key
/// 分页数量
/// 上一个索引
/// 上一个标识
/// 排序方式
///
private async Task<(List 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(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
);
}
///
/// 批量获取指定分页的数据
///
///
/// Hash表缓存key
/// Hash表字段集合
///
private async Task> BatchGetData(
string redisHashCacheKey,
IEnumerable members)
where T : DeviceCacheBasicModel
{
using var pipe = Instance.StartPipe();
foreach (var member in members)
{
pipe.HGet(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));
}
///
/// 处理下一个分页数据
///
///
///
///
///
///
///
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);
}
///
/// 获取下一页游标
///
/// 排序索引ZSET缓存Key
/// 最后一个唯一标识
/// 排序方式
///
private async Task 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
}
///
/// 获取指定ZSET区间内的总数量
///
///
///
///
///
public async Task GetCount(string zsetKey, long min, long max)
{
// 缓存计数优化
var cacheKey = $"{zsetKey}_count_{min}_{max}";
var cached = await Instance.GetAsync(cacheKey);
if (cached.HasValue)
return cached.Value;
var count = await Instance.ZCountAsync(zsetKey, min, max);
await Instance.SetExAsync(cacheKey, 60, count); // 缓存60秒
return count;
}
///
/// 获取指定ZSET的总数量
///
///
///
private async Task GetTotalCount(string redisZSetScoresIndexCacheKey)
{
// 缓存计数优化
var cacheKey = $"{redisZSetScoresIndexCacheKey}_total_count";
var cached = await Instance.GetAsync(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;
}
}
}