track/src/SJZY.Track.Application/Caching/BasicCacheForPeriodicRefresh.cs

516 lines
17 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

using System.Collections.Concurrent;
using FreeRedis;
using Microsoft.Extensions.Caching.Memory;
using Newtonsoft.Json;
using SJZY.Track.Domain.Extensions;
namespace SJZY.Track.Application.Caching;
/// <summary>
/// 容忍短暂的数据不一致,固定周期内强制刷新(本地缓存过期时强制刷新)
/// 适合存字典数据,
/// 注意:缓存过多的大对象有可能OOM,尽量避免
/// 注意:redis内数据只新增/修改,不删!!!
/// </summary>
public abstract class BasicCacheForPeriodicRefresh<TKey, TValue>
where TKey : notnull
where TValue : class
{
private readonly IMemoryCache _memoryCache;
private readonly IRedisClient _redis;
protected BasicCacheForPeriodicRefresh(IRedisClient redis, IMemoryCache? memoryCache = null)
{
_redis = redis;
_memoryCache = memoryCache ?? new MemoryCache(new MemoryCacheOptions
{
CompactionPercentage = 0.75
});
}
protected abstract string BuildRedisKey(TKey key);
protected abstract TimeSpan RedisExpire { get; }
protected virtual TimeSpan LocalExpire => RedisExpire.Divide(2);
protected abstract TimeSpan MissingKeySetNullRedisExpire { get; }
protected virtual TimeSpan MissingKeySetNullLocalExpire => MissingKeySetNullRedisExpire.Divide(2);
protected abstract ValueTask<TValue?> LoadAsync(TKey key);
protected abstract ValueTask<Dictionary<TKey, TValue>> LoadAsync(HashSet<TKey> keys);
private const string ConcurrentKey = "ConcurrentKey";
private readonly ConcurrentDictionary<string, SemaphoreSlim> _semaphoreDict = new()
{
[ConcurrentKey] = new SemaphoreSlim(10, 10)
};
/// <summary>
/// 安全获取缓存
/// 同一个key最多允许并行1个访问
/// </summary>
/// <param name="key"></param>
/// <returns></returns>
public async ValueTask<TValue?> SafeGetAsync(TKey key)
{
var k = key.ToString()!;
var semaphore = _semaphoreDict.GetOrAdd(k, _ => new SemaphoreSlim(1, 1));
await semaphore.WaitAsync();
try
{
return await GetAsync(key);
}
finally
{
semaphore.Release();
_semaphoreDict.TryRemove(k, out _);
}
}
/// <summary>
/// 批量获取缓存
/// 同时最多同时允许10个并发批量缓存访问
/// </summary>
/// <param name="keys"></param>
/// <returns></returns>
public async ValueTask<IReadOnlyDictionary<TKey, TValue>> SafeGetBatchAsync(HashSet<TKey> keys)
{
var semaphore = _semaphoreDict[ConcurrentKey];
await semaphore.WaitAsync();
try
{
return await GetBatchAsync(keys);
}
finally
{
semaphore.Release();
}
}
private async ValueTask<TValue?> GetAsync(TKey key)
{
// 1.LocalLoad
var localEntry = GetLocal(key);
if (localEntry != null)
{
// 1.1 expire refresh for db
if (localEntry.HasExpire)
return await RefreshForDbAsync(key);
// 1.2 not expire return
return localEntry.Data;
}
// 2.RedisLoad
var redisEntry = await GetRedisAsync(key);
if (redisEntry != null)
{
SetLocal(key, redisEntry, redisEntry.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
return redisEntry.Data;
}
// 3.DbLoad
var dbValue = await LoadAsync(key);
if (dbValue != null)
{
SetLocal(key, CacheEntry.Create(dbValue, LocalExpire));
await SetRedisAsync(key, dbValue, LocalExpire);
return dbValue;
}
// db miss set null short expire
SetLocal(key, CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
await SetRedisAsync(key, null, cacheExpire: MissingKeySetNullRedisExpire);
return null;
}
private async ValueTask<TValue?> RefreshForDbAsync(TKey key)
{
// 1.从数据库加载
var dbValue = await LoadAsync(key);
if (dbValue != null)
{
SetLocal(key, CacheEntry.Create(dbValue, LocalExpire));
await SetRedisAsync(key, dbValue);
}
// 2.db加载不到,触发兜底
var redisValue = await GetRedisAsync(key);
if (redisValue != null)
{
// 2.1 设置本地缓存,如果兜底的值为null则需要设置较短过期时间
SetLocal(key, redisValue, redisValue.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
return redisValue.Data;
}
// 3.触发兜底也失败了,设置本地缓存和redis
SetLocal(key, CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
await SetRedisAsync(key, null, cacheExpire: MissingKeySetNullRedisExpire);
return null;
}
private async ValueTask<Dictionary<TKey, TValue>> OnlyLocalMissKeyAsync(List<TKey> keys)
{
var resultDict = new Dictionary<TKey, TValue>();
// 1.redis加载
var redisMissKeys = new HashSet<TKey>();
var localMissKeyArray = keys.ToArray();
var redisValues = await MGetRedisAsync(keys);
for (var i = 0; i < redisValues.Length; i++)
{
var entry = redisValues[i];
if (entry == null)
{
redisMissKeys.Add(localMissKeyArray[i]);
continue;
}
if (entry.Data != null)
resultDict.Add(localMissKeyArray[i], entry.Data);
// 设置本地缓存,空值和非空值过期时间不同
SetLocal(localMissKeyArray[i], entry, entry.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
}
if (redisMissKeys.Count == 0)
return resultDict;
// 2.db加载
var dbValueDict = await LoadAsync(redisMissKeys);
if (dbValueDict.Count > 0)
{
foreach (var kv in dbValueDict)
resultDict.Add(kv.Key, kv.Value);
MSetLocal(dbValueDict!, LocalExpire);
await MSetRedisAsync(dbValueDict!, cacheExpire: RedisExpire);
}
if (dbValueDict.Count == redisMissKeys.Count)
return resultDict;
// 3.db加载失败,设置null值
var dbMissKeys = redisMissKeys.Except(dbValueDict.Keys).ToDictionary(x => x, _ => CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
MSetLocal(dbMissKeys, MissingKeySetNullLocalExpire);
await MSetRedisAsync(dbMissKeys, cacheExpire: MissingKeySetNullRedisExpire);
return resultDict;
}
private async ValueTask<Dictionary<TKey, TValue>> OnlyLocalExpireKeyAsync(List<TKey> keys)
{
var resultDict = new Dictionary<TKey, TValue>();
// 1.DbLoad
var dbValueDict = await LoadAsync(keys.ToHashSet());
foreach (var kv in dbValueDict)
resultDict.Add(kv.Key, kv.Value);
if (dbValueDict.Count == keys.Count)
{
// 刷新本地缓存,还有redis缓存
MSetLocal(dbValueDict!, LocalExpire);
await MSetRedisAsync(dbValueDict!, RedisExpire);
return resultDict;
}
var nullValueSetDict = new Dictionary<TKey, CacheEntry>();
// 2.Redis兜底
var dbMissKeys = keys.Except(dbValueDict.Keys).ToList();
var dbMissKeyArray = dbMissKeys.ToArray();
var redisValues = await MGetRedisAsync(dbMissKeys);
for (var i = 0; i < redisValues.Length; i++)
{
var entry = redisValues[i];
if (entry == null) // 触发兜底也失败了
{
nullValueSetDict.Add(dbMissKeyArray[i], CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
continue;
}
if (entry.Data != null)
resultDict.Add(dbMissKeyArray[i], entry.Data);
// 加载成功设置本地缓存
SetLocal(dbMissKeyArray[i], entry, entry.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
}
// 缓存null值防止缓存穿透
if (nullValueSetDict.Count > 0)
{
MSetLocal(nullValueSetDict, MissingKeySetNullLocalExpire);
await MSetRedisAsync(nullValueSetDict, cacheExpire: MissingKeySetNullRedisExpire);
}
return resultDict;
}
private async ValueTask<IReadOnlyDictionary<TKey, TValue>> GetBatchAsync(HashSet<TKey> keys)
{
if (keys.Count == 0) return Empty<TKey, TValue>.Dict;
// 1.LocalLoad
var localEntries = MGetLocal(keys);
// 选出过期的key和未命中的key
var expireKeys = localEntries.Where(x => x.Value.HasExpire).Select(x => x.Key).ToList();
var localMissKeys = keys.Except(localEntries.Keys).ToList();
// 选出未过期并且不为空的值作为返回的基本组成
var resultDict = localEntries.Where(x => !x.Value.HasExpire && x.Value.Data != null).ToDictionary(x => x.Key, y => y.Value.Data!);
// 不存在过期的也不存在未命中的直接返回即可
if (expireKeys.Count == 0 && localMissKeys.Count == 0)
{
return resultDict;
}
// 只存在未命中的key
if (expireKeys.Count == 0 && localMissKeys.Count > 0)
{
var loadDict = await OnlyLocalMissKeyAsync(localMissKeys);
foreach (var kv in loadDict)
resultDict.Add(kv.Key, kv.Value);
return resultDict;
}
// 只存在过期的key
if (expireKeys.Count > 0 && localMissKeys.Count == 0)
{
var loadDict = await OnlyLocalExpireKeyAsync(expireKeys);
foreach (var kv in loadDict)
resultDict.Add(kv.Key, kv.Value);
return resultDict;
}
// 既存在未命中的key,也存在过期的key
// 2.RedisLoad 仅对missKeys尝试进行redis加载
var redisMissKeys = new List<TKey>();
var localMissKeyArray = localMissKeys.ToArray();
var redisValues = await MGetRedisAsync(localMissKeys);
for (var i = 0; i < redisValues.Length; i++)
{
var entry = redisValues[i];
if (entry == null)
{
redisMissKeys.Add(localMissKeyArray[i]);
continue;
}
if (entry.Data != null)
resultDict.Add(localMissKeyArray[i], entry.Data);
// 加载成功设置本地缓存
SetLocal(localMissKeyArray[i], entry, entry.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
}
// 3.DbLoad redis未命中和过期的key
var unionKeys = new HashSet<TKey>(redisMissKeys.Count + expireKeys.Count);
unionKeys.UnionWith(redisMissKeys);
unionKeys.UnionWith(expireKeys);
var dbValueDict = await LoadAsync(unionKeys);
// 比较新鲜的数据,设置redis缓存+本地缓存
if (dbValueDict.Count > 0)
{
foreach (var kv in dbValueDict)
resultDict.Add(kv.Key, kv.Value);
MSetLocal(dbValueDict!, LocalExpire);
await MSetRedisAsync(dbValueDict!, RedisExpire);
}
// 全部加载成功
if (dbValueDict.Count == unionKeys.Count)
{
return resultDict;
}
var nullValueSetDict = new Dictionary<TKey, CacheEntry>();
// 本地未命中的key,从redis和db加载都失败,稍后将这些key设置一个null值,以防缓存穿透
var dbMissingKeys = redisMissKeys.Except(dbValueDict.Keys).ToList();
if (dbMissingKeys.Count > 0)
{
foreach (var key in dbMissingKeys)
nullValueSetDict.Add(key, CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
}
// 本地过期的key,从db加载失败,触发兜底策略,兜底也失败则将这些key设置一个null值,以防缓存穿透
var expireKeysNotInDb = expireKeys.Except(dbValueDict.Keys).ToList();
if (expireKeysNotInDb.Count > 0)
{
redisValues = await MGetRedisAsync(expireKeysNotInDb);
for (var i = 0; i < redisValues.Length; i++)
{
var entry = redisValues[i];
if (entry == null) // 触发兜底也失败了
{
nullValueSetDict.Add(expireKeysNotInDb[i], CacheEntry.CreateEmpty(MissingKeySetNullLocalExpire));
continue;
}
if (entry.Data != null)
resultDict.Add(localMissKeyArray[i], entry.Data);
// 加载成功设置本地缓存
SetLocal(localMissKeyArray[i], entry, entry.Data == null
? MissingKeySetNullLocalExpire
: LocalExpire
);
}
}
// 缓存null值防止缓存穿透
if (nullValueSetDict.Count > 0)
{
MSetLocal(nullValueSetDict, MissingKeySetNullLocalExpire);
await MSetRedisAsync(nullValueSetDict, MissingKeySetNullRedisExpire);
}
return resultDict;
}
#region 缓存操作方法
private async ValueTask<CacheEntry?> GetRedisAsync(TKey key)
{
return await _redis.GetAsync<CacheEntry?>(BuildRedisKey(key));
}
private async ValueTask<CacheEntry?[]> MGetRedisAsync(List<TKey> keys)
{
var redisKeys = keys.Select(BuildRedisKey).ToArray();
return await _redis.MGetAsync<CacheEntry?>(redisKeys);
}
private CacheEntry? GetLocal(TKey key)
{
return _memoryCache.Get<CacheEntry>(key);
}
private Dictionary<TKey, CacheEntry> MGetLocal(HashSet<TKey> keys)
{
var result = new Dictionary<TKey, CacheEntry>();
foreach (var k in keys)
{
if (_memoryCache.TryGetValue<CacheEntry>(k, out var entry))
result.Add(k, entry!);
}
return result;
}
private Task MSetRedisAsync(Dictionary<TKey, TValue?> dict, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
return MSetRedisAsync(dict.ToDictionary(x => x.Key, y => CacheEntry.Create(y.Value, expire)), cacheExpire);
}
private async Task MSetRedisAsync(Dictionary<TKey, CacheEntry> dict, TimeSpan? cacheExpire = null)
{
var setDict = dict.ToDictionary(x => BuildRedisKey(x.Key), y => y.Value);
await _redis.MSetAsync(setDict);
if (!cacheExpire.HasValue || cacheExpire.Value.TotalSeconds <= 0.0d) return;
// 设置过期时间
foreach (var k in setDict.Keys)
await _redis.ExpireAsync(k, cacheExpire.Value);
}
private Task SetRedisAsync(TKey key, TValue? value, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
cacheExpire = cacheExpire?.TotalSeconds > 0.0 ? cacheExpire : TimeSpan.FromSeconds(-1);
return _redis.SetAsync(BuildRedisKey(key), CacheEntry.Create(value, expire), cacheExpire.Value);
}
private void MSetLocal(Dictionary<TKey, TValue?> dict, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
MSetLocal(dict.ToDictionary(x => x.Key, y => CacheEntry.Create(y.Value, expire)), null, cacheExpire);
}
private void MSetLocal(Dictionary<TKey, CacheEntry> dict, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
foreach (var kv in dict)
SetLocal(kv.Key, kv.Value, expire, cacheExpire);
}
private void SetLocal(TKey key, TValue? value, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
SetLocal(key, CacheEntry.Create(value, expire), null, cacheExpire);
}
private void SetLocal(TKey key, CacheEntry entry, TimeSpan? expire = null, TimeSpan? cacheExpire = null)
{
if (expire.HasValue)
{
entry.SetExpire(expire.Value);
}
_memoryCache.Set(key, entry, cacheExpire?.TotalSeconds > 0.0d
? new MemoryCacheEntryOptions
{
AbsoluteExpirationRelativeToNow = cacheExpire
}
: null
);
}
#endregion
private class CacheEntry
{
private static readonly DateTime NoExpire = DateTime.MaxValue;
public TValue? Data { get; set; }
[System.Text.Json.Serialization.JsonIgnore]
[JsonIgnore]
public DateTime Expire { get; set; }
[System.Text.Json.Serialization.JsonIgnore]
[JsonIgnore]
public bool HasExpire => Expire < DateTime.Now;
public static CacheEntry Create(TValue? data, TimeSpan? expire = null)
{
return new CacheEntry
{
Data = data,
Expire = expire.HasValue ? DateTime.Now.Add(expire.Value) : NoExpire
};
}
public static CacheEntry CreateEmpty(TimeSpan expire)
{
return new CacheEntry
{
Data = null,
Expire = DateTime.Now.Add(expire)
};
}
public void SetExpire(TimeSpan expire)
{
Expire = DateTime.Now.Add(expire);
}
}
}