Merge branch 'S07' into test

This commit is contained in:
hardstylewyl 2025-04-25 11:57:14 +08:00
commit 37e9fd1d91
9 changed files with 769 additions and 140 deletions

View File

@ -0,0 +1,515 @@
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);
}
}
}

View File

@ -0,0 +1,45 @@
using SJZY.Expand.ABP.Core;
using SJZY.Track.Domain.Extensions;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Caching;
/// <summary>
/// 客户id简称缓存
/// </summary>
public class CustomAbbreviationDictCache : BasicCacheForPeriodicRefresh<int, string>, ISingletonDependency
{
private readonly IVExternalSubjectQueries _vExternalSubjectQueries;
public CustomAbbreviationDictCache(IVExternalSubjectQueries vExternalSubjectQueries) : base(RedisHelper.Client)
{
_vExternalSubjectQueries = vExternalSubjectQueries;
}
protected override string BuildRedisKey(int key) => $"{GlobalConst.TrackDataDictKeyPrefix}:CustomAbbreviation:{key}";
protected override TimeSpan RedisExpire => TimeSpan.FromSeconds(-1); // 设置缓存不过期,指定id客户被删除时仍能兜底提供简称
protected override TimeSpan LocalExpire => TimeSpan.FromMinutes(5); // 本地缓存5分钟过期后在次请求会刷新数据
protected override TimeSpan MissingKeySetNullRedisExpire => TimeSpan.FromMinutes(1); // 缓存不存在时设置redisNull
protected override async ValueTask<string?> LoadAsync(int key)
{
if (key == 0) return null;
return await _vExternalSubjectQueries.GetFirstAsync(x => x.Id == key, x => x.Abbreviation);
}
protected override async ValueTask<Dictionary<int, string>> LoadAsync(HashSet<int> keys)
{
if (keys.Count == 0) return Empty<int, string>.Dict;
var result = await _vExternalSubjectQueries.GetListAsync(x => keys.Contains(x.Id), x => new
{
x.Id,
x.Abbreviation
});
return result.Count == 0
? Empty<int, string>.Dict
: result.ToDictionary(x => x.Id, y => y.Abbreviation);
}
}

View File

@ -0,0 +1,46 @@
using SJZY.Expand.ABP.Core;
using SJZY.Track.Domain.Extensions;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Caching;
/// <summary>
/// 系统id中文名称字典缓存
/// </summary>
public class SystemCnNameDictCache : BasicCacheForPeriodicRefresh<int, string>, ISingletonDependency
{
private readonly IVSystemQueries _vSystemQueries;
public SystemCnNameDictCache(IVSystemQueries vSystemQueries) : base(RedisHelper.Client)
{
_vSystemQueries = vSystemQueries;
}
protected override string BuildRedisKey(int key) => $"{GlobalConst.TrackDataDictKeyPrefix}:SystemCnName:{key}";
protected override TimeSpan RedisExpire => TimeSpan.FromSeconds(-1); // 设置缓存不过期,指定id系统被删除时仍能兜底提供系统中文名
protected override TimeSpan LocalExpire => TimeSpan.FromMinutes(10); // 本地缓存10分钟过期后在次请求会刷新数据
protected override TimeSpan MissingKeySetNullRedisExpire => TimeSpan.FromMinutes(1); // 缓存不存在时设置redisNull
protected override async ValueTask<string?> LoadAsync(int key)
{
if (key == 0) return null;
return await _vSystemQueries.GetFirstAsync(x => x.Id == key, x => x.CnName);
}
protected override async ValueTask<Dictionary<int, string>> LoadAsync(HashSet<int> keys)
{
if (keys.Count == 0) return Empty<int, string>.Dict;
var result = await _vSystemQueries.GetListAsync(x => keys.Contains(x.Id), x => new
{
x.Id,
x.CnName
});
return result.Count == 0
? Empty<int, string>.Dict
: result.ToDictionary(x => x.Id, y => y.CnName);
}
}

View File

@ -0,0 +1,45 @@
using SJZY.Expand.ABP.Core;
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Extensions;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Caching;
/// <summary>
/// 轨迹状态id名称字典缓存
/// </summary>
public class TrackStatusNameDictCache : BasicCacheForPeriodicRefresh<int, string>, ISingletonDependency
{
private readonly ITrackStatusRepository _trackStatusRepository;
public TrackStatusNameDictCache(ITrackStatusRepository trackStatusRepository) : base(RedisHelper.Client)
{
_trackStatusRepository = trackStatusRepository;
}
protected override string BuildRedisKey(int key) => $"{GlobalConst.TrackDataDictKeyPrefix}:TrackStatus:{key}";
protected override TimeSpan RedisExpire => TimeSpan.FromSeconds(-1); // 设置缓存不过期,指定id状态删除时仍能提供兜底状态名称
protected override TimeSpan LocalExpire => TimeSpan.FromMinutes(5); // 本地缓存5分钟过期后在次请求会刷新数据
protected override TimeSpan MissingKeySetNullRedisExpire => TimeSpan.FromMinutes(1); // 缓存不存在时设置redisNull
protected override async ValueTask<string?> LoadAsync(int key)
{
if (key == 0) return null;
return await _trackStatusRepository.GetFirstAsync(x => x.Id == key, x => x.Status);
}
protected override async ValueTask<Dictionary<int, string>> LoadAsync(HashSet<int> keys)
{
if (keys.Count == 0) return Empty<int, string>.Dict;
var result = await _trackStatusRepository.GetListAsync(x => keys.Contains(x.Id), x => new
{
x.Id,
x.Status
});
return result.Count == 0
? Empty<int, string>.Dict
: result.ToDictionary(x => x.Id, y => y.Status);
}
}

View File

@ -1,5 +1,5 @@
using Nest;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Caching;
using SJZY.Track.Application.Contract.Dtos.Warning;
using SJZY.Track.Application.Contract.Dtos.Warning.Complete;
using SJZY.Track.Application.Contract.Dtos.Warning.Return;
@ -8,13 +8,11 @@ using SJZY.Track.Application.Services.Warning.Handlers;
using SJZY.Track.Application.Services.Warning.Model;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Aggregates.WarningAggregate;
using SJZY.Track.Domain.Aggregates.WarningAggregate.Complete;
using SJZY.Track.Domain.Dtos.Warning;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Extensions;
using SJZY.Track.Domain.Queries;
namespace SJZY.Track.Application.Services.Warning;
@ -26,26 +24,36 @@ namespace SJZY.Track.Application.Services.Warning;
public class CompleteWarningService : ApplicationService
{
private readonly IOrderRepository _orderRepository;
private readonly IVExternalSubjectQueries _vExternalSubjectQueries;
private readonly IVSystemQueries _vSystemQueries;
private readonly IWarningRuleRepository _warningRuleRepository;
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
private readonly ICompleteWarningHandler _completeWarningHandler;
private readonly ITrackStatusRepository _trackStatusRepository;
private readonly TrackStatusNameDictCache _trackStatusNameDictCache;
private readonly CustomAbbreviationDictCache _customAbbreviationDictCache;
private readonly SystemCnNameDictCache _systemCnNameDictCache;
public CompleteWarningService(IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, ITrackDetailsEsRepository trackDetailsEsRepository, ICompleteWarningHandler completeWarningHandler, ITrackStatusRepository trackStatusRepository, IOrderRepository orderRepository, IVSystemQueries vSystemQueries, IVExternalSubjectQueries vExternalSubjectQueries)
public CompleteWarningService(
IOrderRepository orderRepository,
IWarningRuleRepository warningRuleRepository,
IWarningDetailsEsRepository warningDetailsEsRepository,
ITrackDetailsEsRepository trackDetailsEsRepository,
ICompleteWarningHandler completeWarningHandler,
TrackStatusNameDictCache trackStatusNameDictCache,
CustomAbbreviationDictCache customAbbreviationDictCache,
SystemCnNameDictCache systemCnNameDictCache
)
{
_orderRepository = orderRepository;
_warningRuleRepository = warningRuleRepository;
_warningDetailsEsRepository = warningDetailsEsRepository;
_trackDetailsEsRepository = trackDetailsEsRepository;
_completeWarningHandler = completeWarningHandler;
_trackStatusRepository = trackStatusRepository;
_orderRepository = orderRepository;
_vSystemQueries = vSystemQueries;
_vExternalSubjectQueries = vExternalSubjectQueries;
_trackStatusNameDictCache = trackStatusNameDictCache;
_customAbbreviationDictCache = customAbbreviationDictCache;
_systemCnNameDictCache = systemCnNameDictCache;
}
/// <summary>
/// 获取完整预警规则列表
/// </summary>
@ -58,7 +66,6 @@ public class CompleteWarningService : ApplicationService
if (result.Count == 0)
return new UnifyResultDto<List<CompleteWarningRuleItemDto>>(true, new List<CompleteWarningRuleItemDto>());
// todo:缓存优化
var trackStatusIds = result
.SelectMany(x =>
{
@ -72,13 +79,9 @@ public class CompleteWarningService : ApplicationService
return requireStatusIds.SelectMany(y => y);
}).ToHashSet();
var trackStatusDict = (await _trackStatusRepository.GetListAsync(x => trackStatusIds.Contains(x.Id), x => new
{
x.Id,
x.Status
})).ToDictionary(x => x.Id, y => y.Status);
// 查订单类型映射
// 查轨迹状态名称字典
var trackStatusDict = await _trackStatusNameDictCache.SafeGetBatchAsync(trackStatusIds);
// 查订单类型字典
var orderTypeDict = await _orderRepository.GetTrackOrderTypeDictAsync();
var map = result.Select(x => new CompleteWarningRuleItemDto
@ -136,40 +139,27 @@ public class CompleteWarningService : ApplicationService
result.Pagination.Total);
}
// todo:缓存优化
// 订单类型名称字典
var orderTypeDict = await _orderRepository.GetTrackOrderTypeDictAsync();
var trackStatusIds = result.List
.SelectMany(x => x.GetMissingTrackStatusIdGroups().SelectMany(y => y))
.ToHashSet();
var trackStatusDict = (await _trackStatusRepository.GetListAsync(x => trackStatusIds.Contains(x.Id), x => new
{
x.Id,
x.Status
})).ToDictionary(x => x.Id, y => y.Status);
// 轨迹状态名称字典
var trackStatusDict = await _trackStatusNameDictCache.SafeGetBatchAsync(trackStatusIds);
// 客户id简称字典
var customIds = result.List.Select(x => x.CustomId).ToHashSet();
var customDict = await _customAbbreviationDictCache.SafeGetBatchAsync(customIds);
// 子系统id中文名字典
var subSystemIds = result.List.Select(x => x.SubSystemId).ToHashSet();
var subSystemDict = await _systemCnNameDictCache.SafeGetBatchAsync(subSystemIds);
// 查最新轨迹(创建时间)
var orderIds = result.List.Select(x => x.OrderId).ToHashSet();
var orderIdTrackDict = await _trackDetailsEsRepository.GetLatestDictAsync(orderIds);
// todo:缓存优化
// 订单类型
var orderTypeDict = await _orderRepository.GetTrackOrderTypeDictAsync();
// 客户简称
var customIds = result.List.Select(x => x.CustomId).ToHashSet();
var customDict = (await _vExternalSubjectQueries
.GetListAsync(x => customIds.Contains(x.Id), x => new
{
x.Id,
x.Abbreviation
})).ToDictionary(x => x.Id, x => x.Abbreviation);
// 子系统
var subSystemIds = result.List.Select(x => x.SubSystemId).ToHashSet();
var subSystemDict = (await _vSystemQueries
.GetListAsync(x => subSystemIds.Contains(x.Id), x => new
{
x.Id,
x.CnName
})).ToDictionary(x => x.Id, x => x.CnName);
var map = result.List.Select(x =>
{
var orderIdTrack = orderIdTrackDict.GetValueOrDefault(x.OrderId);

View File

@ -1,5 +1,6 @@
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Caching;
using SJZY.Track.Application.Contract.Dtos.Warning;
using SJZY.Track.Application.Contract.Dtos.Warning.Return;
using SJZY.Track.Application.Contract.Events;
@ -7,7 +8,6 @@ using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Services.Warning.Handlers;
using SJZY.Track.Application.Services.Warning.Model;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Aggregates.WarningAggregate;
using SJZY.Track.Domain.Dtos.Warning;
using SJZY.Track.Domain.Enums;
@ -25,27 +25,39 @@ namespace SJZY.Track.Application.Services.Warning;
public class ReturnWarningService : ApplicationService
{
private readonly IOrderRepository _orderRepository;
private readonly IVExternalSubjectQueries _vExternalSubjectQueries;
private readonly IVSystemQueries _vSystemQueries;
private readonly IReturnWarningHandler _returnWarningHandler;
private readonly IWarningRuleRepository _warningRuleRepository;
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
private readonly ITrackStatusRepository _trackStatusRepository;
private readonly ICapPublisher _capPublisher;
private readonly IHttpClientProxy<IUserBaseInfoService> _requestCurrentUserService;
private readonly TrackStatusNameDictCache _trackStatusNameDictCache;
private readonly CustomAbbreviationDictCache _customAbbreviationDictCache;
private readonly SystemCnNameDictCache _systemCnNameDictCache;
public ReturnWarningService(IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IReturnWarningHandler returnWarningHandler, ITrackStatusRepository trackStatusRepository, IOrderRepository orderRepository, IVExternalSubjectQueries vExternalSubjectQueries, IVSystemQueries vSystemQueries, ICapPublisher capPublisher, IHttpClientProxy<IUserBaseInfoService> requestCurrentUserService)
public ReturnWarningService(
IOrderRepository orderRepository,
IVSystemQueries vSystemQueries,
IReturnWarningHandler returnWarningHandler,
IWarningRuleRepository warningRuleRepository,
IWarningDetailsEsRepository warningDetailsEsRepository,
ICapPublisher capPublisher,
IHttpClientProxy<IUserBaseInfoService> requestCurrentUserService,
TrackStatusNameDictCache trackStatusNameDictCache,
CustomAbbreviationDictCache customAbbreviationDictCache,
SystemCnNameDictCache systemCnNameDictCache
)
{
_orderRepository = orderRepository;
_vSystemQueries = vSystemQueries;
_returnWarningHandler = returnWarningHandler;
_warningRuleRepository = warningRuleRepository;
_warningDetailsEsRepository = warningDetailsEsRepository;
_returnWarningHandler = returnWarningHandler;
_trackStatusRepository = trackStatusRepository;
_orderRepository = orderRepository;
_vExternalSubjectQueries = vExternalSubjectQueries;
_vSystemQueries = vSystemQueries;
_capPublisher = capPublisher;
_requestCurrentUserService = requestCurrentUserService;
_trackStatusNameDictCache = trackStatusNameDictCache;
_customAbbreviationDictCache = customAbbreviationDictCache;
_systemCnNameDictCache = systemCnNameDictCache;
}
/// <summary>
@ -61,16 +73,10 @@ public class ReturnWarningService : ApplicationService
if (result.Count == 0)
return new UnifyResultDto<List<ReturnWarningRuleItemDto>>(true, new List<ReturnWarningRuleItemDto>());
// todo:缓存优化
var trackStatusIds = result
.SelectMany(x => x.TrackStatusIds.Split(',').Select(int.Parse))
.ToHashSet();
var trackStatusDict = (await _trackStatusRepository.GetListAsync(x => trackStatusIds.Contains(x.Id), x => new
{
x.Id,
x.Status
})).ToDictionary(x => x.Id, y => y.Status);
var trackStatusDict = await _trackStatusNameDictCache.SafeGetBatchAsync(trackStatusIds);
var map = result.Select(x => new ReturnWarningRuleItemDto
{
@ -114,25 +120,14 @@ public class ReturnWarningService : ApplicationService
result.Pagination.Total);
}
// todo:缓存优化
// 订单类型
// 订单类型名称字典
var orderTypeDict = await _orderRepository.GetTrackOrderTypeDictAsync();
// 客户简称
// 客户id简称字典
var customIds = result.List.Select(x => x.CustomId).ToHashSet();
var customDict = (await _vExternalSubjectQueries
.GetListAsync(x => customIds.Contains(x.Id), x => new
{
x.Id,
x.Abbreviation
})).ToDictionary(x => x.Id, x => x.Abbreviation);
// 子系统
var customDict = await _customAbbreviationDictCache.SafeGetBatchAsync(customIds);
// 子系统id中文名字典
var subSystemIds = result.List.Select(x => x.SubSystemId).ToHashSet();
var subSystemDict = (await _vSystemQueries
.GetListAsync(x => subSystemIds.Contains(x.Id), x => new
{
x.Id,
x.CnName
})).ToDictionary(x => x.Id, x => x.CnName);
var subSystemDict = await _systemCnNameDictCache.SafeGetBatchAsync(subSystemIds);
var map = result.List.Select(x => new ReturnWarningDetailsDto
{
@ -261,7 +256,7 @@ public class ReturnWarningService : ApplicationService
// 1.查出预警明细
var ids = input.WarningDetailIds.ToHashSet();
var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO()
var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO
{
DetailsIds = ids
});
@ -309,7 +304,6 @@ public class ReturnWarningService : ApplicationService
{
[detail.OrderId] = KeyValuePair.Create(detail.TrackEsId, detail.TrackVersion)
},
// OrderIds = new[] { order.Id },
TrackCode = detail.TrackCode,
OperationMan = userInfo.Data.UserName,
SystemPushInfo = new SystemPushStruct

View File

@ -1,4 +1,5 @@
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Caching;
using SJZY.Track.Application.Contract.Dtos.Warning;
using SJZY.Track.Application.Contract.Dtos.Warning.Transport;
using SJZY.Track.Application.Helpers;
@ -7,13 +8,11 @@ using SJZY.Track.Application.Services.Warning.Model;
using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Aggregates.WarningAggregate;
using SJZY.Track.Domain.Aggregates.WarningAggregate.Transport;
using SJZY.Track.Domain.Dtos.Warning;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Extensions;
using SJZY.Track.Domain.Queries;
namespace SJZY.Track.Application.Services.Warning;
@ -25,26 +24,36 @@ namespace SJZY.Track.Application.Services.Warning;
public class TransportWarningService : ApplicationService
{
private readonly IOrderRepository _orderRepository;
private readonly IVExternalSubjectQueries _vExternalSubjectQueries;
private readonly IVSystemQueries _vSystemQueries;
private readonly IKpiAreaRepository _kpiAreaRepository;
private readonly IWarningRuleRepository _warningRuleRepository;
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
private readonly ITransportWarningHandler _transportWarningHandler;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
private readonly ITrackStatusRepository _trackStatusRepository;
private readonly IKpiAreaRepository _kpiAreaRepository;
private readonly ITransportWarningHandler _transportWarningHandler;
private readonly TrackStatusNameDictCache _trackStatusNameDictCache;
private readonly CustomAbbreviationDictCache _customAbbreviationDictCache;
private readonly SystemCnNameDictCache _systemCnNameDictCache;
public TransportWarningService(IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, ITransportWarningHandler transportWarningHandler, ITrackDetailsEsRepository trackDetailsEsRepository, ITrackStatusRepository trackStatusRepository, IKpiAreaRepository kpiAreaRepository, IOrderRepository orderRepository, IVExternalSubjectQueries vExternalSubjectQueries, IVSystemQueries vSystemQueries)
public TransportWarningService(
IOrderRepository orderRepository,
IKpiAreaRepository kpiAreaRepository,
IWarningRuleRepository warningRuleRepository,
IWarningDetailsEsRepository warningDetailsEsRepository,
ITrackDetailsEsRepository trackDetailsEsRepository,
ITransportWarningHandler transportWarningHandler,
TrackStatusNameDictCache trackStatusNameDictCache,
CustomAbbreviationDictCache customAbbreviationDictCache,
SystemCnNameDictCache systemCnNameDictCache
)
{
_orderRepository = orderRepository;
_kpiAreaRepository = kpiAreaRepository;
_warningRuleRepository = warningRuleRepository;
_warningDetailsEsRepository = warningDetailsEsRepository;
_transportWarningHandler = transportWarningHandler;
_trackDetailsEsRepository = trackDetailsEsRepository;
_trackStatusRepository = trackStatusRepository;
_kpiAreaRepository = kpiAreaRepository;
_orderRepository = orderRepository;
_vExternalSubjectQueries = vExternalSubjectQueries;
_vSystemQueries = vSystemQueries;
_transportWarningHandler = transportWarningHandler;
_trackStatusNameDictCache = trackStatusNameDictCache;
_customAbbreviationDictCache = customAbbreviationDictCache;
_systemCnNameDictCache = systemCnNameDictCache;
}
/// <summary>
@ -60,11 +69,10 @@ public class TransportWarningService : ApplicationService
var areaIds = entities
.Select(x => x.AreaId)
.Where(x => x != null)
.Where(x => x > 0)
.Select(x => x!.Value)
.ToHashSet();
// todo:缓存优化
Dictionary<int, string>? areaNameDict = null;
if (areaIds.Count > 0)
{
@ -136,17 +144,12 @@ public class TransportWarningService : ApplicationService
{
var entities = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(input.WarningId, input.SceneId);
// todo:缓存优化
var trackStatusIds = entities.SelectMany(x => new List<int>(2)
{
x.StartTrackStatusId,
x.EndTrackStatusId
}).ToHashSet();
var trackStatusDict = (await _trackStatusRepository.GetListAsync(x => trackStatusIds.Contains(x.Id), x => new
{
x.Id,
x.Status
})).ToDictionary(x => x.Id, y => y.Status);
var trackStatusDict = await _trackStatusNameDictCache.SafeGetBatchAsync(trackStatusIds);
var map = entities.Select(x => new TransportWarningSceneRuleItemDto
{
@ -190,25 +193,14 @@ public class TransportWarningService : ApplicationService
// 查轨迹状态最早轨迹
var orderIdTrackStatusEarliestTrackDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
// todo:缓存优化
// 订单类型
// 订单类型名称字典
var orderTypeDict = await _orderRepository.GetTrackOrderTypeDictAsync();
// 客户简称
// 客户id简称字典
var customIds = result.List.Select(x => x.CustomId).ToHashSet();
var customDict = (await _vExternalSubjectQueries
.GetListAsync(x => customIds.Contains(x.Id), x => new
{
x.Id,
x.Abbreviation
})).ToDictionary(x => x.Id, x => x.Abbreviation);
// 子系统
var customDict = await _customAbbreviationDictCache.SafeGetBatchAsync(customIds);
// 子系统id中文名字典
var subSystemIds = result.List.Select(x => x.SubSystemId).ToHashSet();
var subSystemDict = (await _vSystemQueries
.GetListAsync(x => subSystemIds.Contains(x.Id), x => new
{
x.Id,
x.CnName
})).ToDictionary(x => x.Id, x => x.CnName);
var subSystemDict = await _systemCnNameDictCache.SafeGetBatchAsync(subSystemIds);
var map = result.List.Select(x =>
{

View File

@ -1,8 +1,8 @@
using SJZY.Track.Application.Caching;
using SJZY.Track.Application.Contract.Dtos.Warning;
using SJZY.Track.Application.Services.Warning.Handlers;
using SJZY.Track.Domain.Aggregates.WarningAggregate;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Queries;
namespace SJZY.Track.Application.Services.Warning;
@ -14,20 +14,27 @@ namespace SJZY.Track.Application.Services.Warning;
public class WarningService : ApplicationService
{
private readonly IWarningRuleRepository _warningRuleRepository;
private readonly IVSystemQueries _vSystemQueries;
private readonly IVExternalSubjectQueries _vExternalSubjectQueries;
private readonly ICompleteWarningHandler _completeWarningHandler;
private readonly ITransportWarningHandler _transportWarningHandler;
private readonly IReturnWarningHandler _returnWarningHandler;
private readonly CustomAbbreviationDictCache _customAbbreviationDictCache;
private readonly SystemCnNameDictCache _systemCnNameDictCache;
public WarningService(IWarningRuleRepository warningRuleRepository, IVSystemQueries vSystemQueries, IVExternalSubjectQueries vExternalSubjectQueries, IReturnWarningHandler returnWarningHandler, ITransportWarningHandler transportWarningHandler, ICompleteWarningHandler completeWarningHandler)
public WarningService(
IWarningRuleRepository warningRuleRepository,
ICompleteWarningHandler completeWarningHandler,
ITransportWarningHandler transportWarningHandler,
IReturnWarningHandler returnWarningHandler,
CustomAbbreviationDictCache customAbbreviationDictCache,
SystemCnNameDictCache systemCnNameDictCache
)
{
_warningRuleRepository = warningRuleRepository;
_vSystemQueries = vSystemQueries;
_vExternalSubjectQueries = vExternalSubjectQueries;
_returnWarningHandler = returnWarningHandler;
_transportWarningHandler = transportWarningHandler;
_completeWarningHandler = completeWarningHandler;
_transportWarningHandler = transportWarningHandler;
_returnWarningHandler = returnWarningHandler;
_customAbbreviationDictCache = customAbbreviationDictCache;
_systemCnNameDictCache = systemCnNameDictCache;
}
/// <summary>
@ -42,22 +49,12 @@ public class WarningService : ApplicationService
if (result.Count == 0)
return new UnifyResultDto<List<WarningRuleDto>>(true, new List<WarningRuleDto>());
// todo:缓存优化
// 客户id简称字典
var customIds = result.Select(x => x.CustomId).ToHashSet();
var customDict = (await _vExternalSubjectQueries
.GetListAsync(x => customIds.Contains(x.Id), x => new
{
x.Id,
x.Abbreviation
})).ToDictionary(x => x.Id, x => x.Abbreviation);
var customDict = await _customAbbreviationDictCache.SafeGetBatchAsync(customIds);
// 子系统id中文名字典
var subSystemIds = result.Select(x => x.SubSystemId).ToHashSet();
var subSystemDict = (await _vSystemQueries
.GetListAsync(x => subSystemIds.Contains(x.Id), x => new
{
x.Id,
x.CnName
})).ToDictionary(x => x.Id, x => x.CnName);
var subSystemDict = await _systemCnNameDictCache.SafeGetBatchAsync(subSystemIds);
return new UnifyResultDto<List<WarningRuleDto>>(true, result.Select(x => new WarningRuleDto
{

View File

@ -91,4 +91,9 @@ public partial class GlobalConst
/// 轨迹中台hangfire key前缀
/// </summary>
public const string TrackHangfireKeyPrefix = "TrackHangfire";
/// <summary>
/// 轨迹中台数据字典key前缀
/// </summary>
public const string TrackDataDictKeyPrefix = "TrackDataDict";
}