feat: 轨迹预警规则缓存以及集群通知

src/SJZY.Track.Api/SjzyTrackApiModule.cs

@@ -9,6 +9,7 @@ using SJZY.Expand.ABP.Core.Prometheus;
 using SJZY.Expand.ABP.Core.Redis;
 using SJZY.Expand.ABP.Core.Swagger;
 using SJZY.Track.Application;
+using SJZY.Track.Application.HostServices;
 using SJZY.Track.Repository;
 using Volo.Abp;
 using Volo.Abp.AspNetCore.Mvc;
@@ -62,6 +63,7 @@ namespace SJZY.Track.Api
             Configure<AbpAuditingOptions>(options => { options.IsEnabled = false; });
 
             context.Services.AddRedis();
+            context.Services.AddHostedService<RedisHostService>();
         }
 
         public override void OnApplicationInitialization(ApplicationInitializationContext context)

src/SJZY.Track.Application.Contract/Events/ModifyWarningEvent.cs

@@ -0,0 +1,23 @@
+namespace SJZY.Track.Application.Contract.Events;
+
+/// <summary>
+/// redis 的发布订阅事件
+/// </summary>
+public class ModifyWarningEvent:RedisEvent
+{
+    public string Key { get; set; }
+}
+
+public class RedisEvent : IRedisEvent
+{
+    public string EventType { get; set; }
+}
+
+
+/// <summary>
+/// 简单的事件接口，后续可以根据需要扩展
+/// </summary>
+public interface IRedisEvent
+{ 
+    string EventType { get; set; }
+}

src/SJZY.Track.Application.Contract/Services/IForewarningCacheService.cs

@@ -19,4 +19,16 @@ public interface IForewarningCacheService
     /// <param name="customerId"></param>
     /// <returns></returns>
     Task DeleteRulesAsync(int subSystemId, int customerId);
+
+    /// <summary>
+    /// 重置本地缓存
+    /// </summary>
+    /// <returns></returns>
+    Task Clear();
+    
+    /// <summary>
+    /// 清除redis 缓存
+    /// </summary>
+    /// <returns></returns>
+    Task ClearRedis();
 }

src/SJZY.Track.Application/HostServices/RedisHostService.cs

@@ -0,0 +1,46 @@
+using Microsoft.Extensions.Hosting;
+using SJZY.Expand.ABP.Core;
+using SJZY.Track.Application.Contract.Events;
+using SJZY.Track.Application.Contract.Services;
+using SJZY.Track.Domain.Consts;
+
+namespace SJZY.Track.Application.HostServices;
+
+public class RedisHostService:IHostedService
+{
+    readonly IForewarningCacheService _forewarningCacheService;
+
+    public RedisHostService(IForewarningCacheService forewarningCacheService)
+    {
+        _forewarningCacheService = forewarningCacheService;
+    }
+
+    public  Task StartAsync(CancellationToken cancellationToken)
+    {
+        // 订阅频道
+        RedisHelper.Client.Subscribe(GetKey(), Subscribe);
+        return Task.CompletedTask;
+    }
+
+    public Task StopAsync(CancellationToken cancellationToken)
+    {
+        RedisHelper.Client.UnSubscribe(GetKey());
+        return Task.CompletedTask;
+    }
+
+    private void Subscribe(string message,object value)
+    {
+        if (value is string)
+        {
+            // 暂时只处理一个事件，不做反序列化
+            //var instance = Newtonsoft.Json.JsonConvert.DeserializeObject<ModifyWarningEvent>(value as string);
+        }
+        
+        _forewarningCacheService.Clear();
+    }
+    
+    private string GetKey()
+    {
+        return $"{GlobalConst.PublishNotifyChannelKey}";
+    }
+}

src/SJZY.Track.Application/Services/ForewarningServices/ForewarningCacheService.cs

@@ -1,29 +1,133 @@
+using FreeRedis;
+using SJZY.Expand.ABP.Core;
+using SJZY.Expand.ABP.Core.PublicMethords;
 using SJZY.Track.Application.Contract.Dtos.Forewarning;
 using SJZY.Track.Application.Contract.Services;
+using SJZY.Track.Domain.Aggregates.ForewarningAggregate;
+using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
+using SJZY.Track.Domain.Consts;
+using SJZY.Track.Domain.Dtos.Forewarning;
 using Volo.Abp.DependencyInjection;
 
 namespace SJZY.Track.Application.Services.ForewarningServices;
 
 public class ForewarningCacheService: IForewarningCacheService, ISingletonDependency
 {
-    private Dictionary<string, List<ForewarningRuleCacheEntry>> _rules = new();
+    readonly IKpiAreaRepository _kpiAreaRepository;
-    string AreaKey(int subSystemId) => $"TrackArea:{subSystemId}";
+    readonly IWarningRepository _warningRepository;
-    string WarningKey(int subSystemId, int customerId) => $"TrackWarning:{subSystemId}:{customerId}";
     
+    const string GlobalKpiAreaLockKey = "KKpiAreaLockKey_GlobalLock";
+    private const string GlobalWaningLockKey = "GlobalWaningLockKey_GlobalLock";
+    
+    private Dictionary<string, List<ForewarningRuleCacheEntry>> _rules = new();
+    private Dictionary<int,List<AreaCacheEntry>>? _areas = new();
+    readonly object _lock = new object();
+
+    public ForewarningCacheService(IKpiAreaRepository kpiAreaRepository, IWarningRepository warningRepository)
+    {
+        _kpiAreaRepository = kpiAreaRepository;
+        _warningRepository = warningRepository;
+    }
+
+
+    string WarningKey(int subSystemId, int customerId) => $"{GlobalConst.TrackWarningKeyPrefix}:{subSystemId}:{customerId}";
+    
+    string WarningKey() => $"{GlobalConst.TrackWarningKeyPrefix}:all";
+    
+    string AreaKey()
+    {
+        return $"{GlobalConst.TrackAreaKeyPrefix}";
+    }
+
     // 30w => 1w => local
     public async Task<List<ForewarningRuleCacheEntry>?> GetRulesAsync(int subSystemId, int customerId)
     {
-        var rules = _rules.GetValueOrDefault(WarningKey(subSystemId, customerId));
+        var key = WarningKey(subSystemId, customerId);
+        var rules = _rules.GetValueOrDefault(key);
 
         if (rules == null)
-        {
+        { 
-            // todo: 写入缓存  set str
+            return await GetRules(subSystemId, customerId);
-            await Task.Delay(1);
         }
 
         return rules;
     }
 
+    private List<ForewarningRuleCacheEntry> WriteRules(List<WarningRuleCacheDto> rules)
+    {
+        List<ForewarningRuleCacheEntry> result = new List<ForewarningRuleCacheEntry>();
+        foreach (var item in rules)
+        {
+            List<AreaCacheEntry>? areaList = null;
+            if (item.AreaId.HasValue)
+            {
+                areaList = GetAreaInfoByAreaId(item.AreaId!.Value);
+            }
+            
+            var entry = new ForewarningRuleCacheEntry
+            {
+                Id = item.Id,
+                WarningId = item.WarningId,
+                TotalTimeSpan = item.TotalTimeSpan,
+                WarnTimeSpan = item.WarnTimeSpan,
+                WarnCount = item.WarnCount,
+                StartTrackStatusId = item.StartTrackStatusId,
+                EndTrackStatusId = item.EndTrackStatusId,
+                CargoType = item.CargoType,
+                Months = string.IsNullOrEmpty(item.Months)?new List<int>():item.Months.Split('/').Select(int.Parse).ToList(),
+                IsTransit = item.IsTransit,
+                AreaId = areaList,
+                ProductCode = item.ProductCode,
+                LeavePort = item.LeavePort
+            };
+            
+            result.Add(entry);
+        }
+
+        return result;
+    }
+
+    private async Task<List<ForewarningRuleCacheEntry>?> GetRules(int systemId, int customerId)
+    {
+        var key = WarningKey();
+        var result = RedisHelper.Client.Get<List<WarningRuleCacheDto>>(key);
+        if (result != null)
+        {
+            // 非线程安全，需要加锁
+            lock (_lock)
+            {
+                _rules = result.GroupBy(val => WarningKey(val.SystemId, val.CustomerId))
+                    .ToDictionary(val => val.Key, val => WriteRules(val.ToList()));
+
+                return _rules.GetValueOrDefault(WarningKey(systemId,customerId));
+            }
+        }
+        
+        // 非线程安全，需要加锁
+        using (RedisHelper.Client.Lock(GlobalWaningLockKey, 20))
+        {
+            var ruleResult = _rules.GetValueOrDefault(WarningKey(systemId,customerId));
+            if (ruleResult == null)
+            {
+                var warnings  = await _warningRepository.GetAllWarningRule();
+                if (warnings == null || warnings.Count == 0)
+                {
+                    RedisHelper.Client.Set(key, new List<WarningRuleCacheDto>(), TimeSpan.FromSeconds(20));
+                    return new List<ForewarningRuleCacheEntry>();
+                }
+                
+                RedisHelper.Client.Set(key, warnings);
+                _rules = warnings.GroupBy(val => WarningKey(val.SystemId, val.CustomerId))
+                    .ToDictionary(val => val.Key, val => WriteRules(val.ToList()));
+
+                return _rules.GetValueOrDefault(WarningKey(systemId,customerId));
+            }
+
+            return ruleResult;
+        }
+
+    }
+    
     // a b c  del local   cusomer 3 del redis cache  pub del redis cache
     public Task DeleteRulesAsync(int subSystemId, int customerId)
     {
@@ -31,4 +135,74 @@ public class ForewarningCacheService: IForewarningCacheService, ISingletonDepend
 
         return Task.CompletedTask;
     }
+
+    private List<AreaCacheEntry>? GetAreaInfoByAreaId(int areaId)
+    {
+        if (_areas == null || _areas.Count == 0)
+        {
+            var areas = GetAreas();
+            if(areas==null || areas.Count==0) return null;
+            _areas = areas.GroupBy(x => x.AreaId)
+                .ToDictionary(x => x.Key, x => x.Select(KpiAreaInfoToEntry).ToList());
+        }
+        
+        return _areas.GetValueOrDefault(areaId);
+    }
+
+    AreaCacheEntry KpiAreaInfoToEntry(KpiAreaInfo model)
+    {
+        return new AreaCacheEntry
+        {
+            Country = model.Country,
+            Province = model.Province,
+            City = model.City,
+            LeftPostcode = model.LeftPostcode,
+            RightPostcode = model.RightPostcode
+        };
+    }
+
+    /// <summary>
+    /// 获取所有区域
+    /// </summary>
+    /// <returns></returns>
+    private List<KpiAreaInfo>? GetAreas()
+    {
+        var key = AreaKey();
+        var result = RedisHelper.Client.Get<List<KpiAreaInfo>>(key);
+        if (result == null)
+        {
+            using (RedisHelper.Client.Lock(GlobalKpiAreaLockKey, 20))
+            {
+                result = RedisHelper.Client.Get<List<KpiAreaInfo>>(key);
+                if (result != null) return new List<KpiAreaInfo>();
+                
+                var kpiArea =  _kpiAreaRepository.GetAllAreaInfos();
+                if (kpiArea == null)
+                {          
+                    RedisHelper.Client.Set(key, new List<KpiAreaInfo>(), TimeSpan.FromSeconds(20));
+                    return new List<KpiAreaInfo>();
+                }
+                
+                RedisHelper.Client.Set(key, kpiArea);
+                return kpiArea;
+            }
+        }
+
+        return result;
+    }
+    
+    public Task Clear()
+    {
+        _rules?.Clear();
+        _areas?.Clear();
+        return Task.CompletedTask;
+    }
+
+    public Task ClearRedis()
+    {
+        RedisHelper.Client.Del(WarningKey());
+        RedisHelper.Client.Del(AreaKey());
+        return Task.CompletedTask;
+    }
+
 }

src/SJZY.Track.Application/Services/ForewarningServices/KpiAreaCacheService.cs

@@ -22,25 +22,8 @@ public class KpiAreaCacheService:IKpiAreaCacheService,ITransientDependency
         return $"{GlobalConst.KpiAreaKeyPrefix}:{systemId}:{country}";
     }
     
-    public async Task GetKpiArea(int systemId, string country)
+    public Task GetKpiArea(int systemId, string country)
     {
-        var key = GetKey(systemId, country);
+        throw new NotImplementedException();
-        var result = RedisHelper.Client.Get<KpiAreaInfo>(key);
-        if (result == null)
-        {
-            using (var globalLock = RedisHelper.Client.Lock(GlobalKpiAreaLockKey, 10))
-            {
-                result = RedisHelper.Client.Get<KpiAreaInfo>(key);
-                if (result != null) return;
-
-                //todo 转换成KpiAreaInfo
-                var kpiArea =await _kpiAreaRepository.GetAllAreaInfos();
-                if (kpiArea == null)
-                {
-                    return;
-                }
-                RedisHelper.Client.Set(key, kpiArea, TimeSpan.FromSeconds(20));
-            }
-        }
     }
 }

src/SJZY.Track.Application/Services/ForewarningServices/KpiAreaService.cs

@@ -1,4 +1,7 @@
+using SJZY.Expand.ABP.Core;
+using SJZY.Track.Application.Contract.Services;
 using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
+using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Dtos.Forewarning;
 
 namespace SJZY.Track.Application.Services.ForewarningServices;
@@ -8,10 +11,12 @@ namespace SJZY.Track.Application.Services.ForewarningServices;
 public class KpiAreaService : ApplicationService
 {
     private readonly IKpiAreaRepository _ruleAreaRepository;
+    private readonly IForewarningCacheService _forewarningCacheService;
 
-    public KpiAreaService(IKpiAreaRepository ruleAreaRepository)
+    public KpiAreaService(IKpiAreaRepository ruleAreaRepository, IForewarningCacheService forewarningCacheService)
     {
         _ruleAreaRepository = ruleAreaRepository;
+        _forewarningCacheService = forewarningCacheService;
     }
 
     /// <summary>
@@ -37,6 +42,8 @@ public class KpiAreaService : ApplicationService
     {
         var ruleAreaEntity = ObjectMapper.Map<AddRuleAreaInfoRequest, RuleAreaInfoEntity>(input);
         var ruleInfoId = await _ruleAreaRepository.AddAreaInfoAsync(ruleAreaEntity);
+        await _forewarningCacheService.ClearRedis();
+        await RedisHelper.Client.PublishAsync(GlobalConst.PublishNotifyChannelKey, "clear local cache");
         return new UnifyResultDto(true, ruleInfoId, "添加成功");
     }
 
@@ -49,6 +56,8 @@ public class KpiAreaService : ApplicationService
     public async ValueTask<UnifyResultDto> DeleteKpiArea(int id)
     {
         await _ruleAreaRepository.DeleteArea(id);
+        await _forewarningCacheService.ClearRedis();
+        await RedisHelper.Client.PublishAsync(GlobalConst.PublishNotifyChannelKey, "clear locl cache");
         return new UnifyResultDto(true, "删除成功");
     }
 
@@ -61,6 +70,8 @@ public class KpiAreaService : ApplicationService
     public async ValueTask<UnifyResultDto> DeleteKpiAreaInfo(int id)
     {
         await _ruleAreaRepository.DeleteAreaInfo(id);
+        await _forewarningCacheService.ClearRedis();
+        await RedisHelper.Client.PublishAsync(GlobalConst.PublishNotifyChannelKey, "clear local cache");
         return new UnifyResultDto(true, "删除成功");
     }
 
@@ -101,6 +112,9 @@ public class KpiAreaService : ApplicationService
 
         var ruleAreaEntity = ObjectMapper.Map(input, areaInfo);
         await _ruleAreaRepository.UpdateAreaInfoAsync(ruleAreaEntity);
+
+        await _forewarningCacheService.ClearRedis();
+        await RedisHelper.Client.PublishAsync(GlobalConst.PublishNotifyChannelKey, "clear local cache");
         return new UnifyResultDto(true, "更新成功");
     }
 

src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRepository.cs

@@ -59,4 +59,11 @@ public interface IWarningRepository : IBasicRepository<WarningEntity>
     /// <param name="warningId"></param>
     /// <returns></returns>
     Task<WarningEntity?> GetByIdAsync(int warningId);
+
+    /// <summary>
+    /// 获取预警规则列表
+    /// </summary>
+    /// <returns></returns>
+    Task<List<WarningRuleCacheDto>?> GetAllWarningRule();
+
 }

src/SJZY.Track.Domain/Aggregates/KPIAreaAggregate/IKpiAreaRepository.cs

@@ -64,5 +64,5 @@ public interface IKpiAreaRepository : IBasicRepository<RuleAreaEntity>
     /// 获取所有区域信息
     /// </summary>
     /// <returns></returns>
-    Task<List<RuleAreaInfoEntity>> GetAllAreaInfos();
+    List<KpiAreaInfo>? GetAllAreaInfos();
 }

src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs

@@ -42,4 +42,19 @@ public partial class GlobalConst
     /// 港口时区redis key前缀
     /// </summary>
     public const string PortTimezoneKeyPrefix = "PortTimezone";
+    
+    /// <summary>
+    /// 发布订阅通知频道Key
+    /// </summary>
+    public const string PublishNotifyChannelKey = "PublishNotifyChannel";
+    
+    /// <summary>
+    /// 轨迹区域reids key
+    /// </summary>
+    public const string TrackAreaKeyPrefix = "TrackArea";
+    
+    /// <summary>
+    /// 轨迹预警数据redis key前缀
+    /// </summary>
+    public const string TrackWarningKeyPrefix = "TrackWarning";
 }

src/SJZY.Track.Domain/Dtos/Forewarning/KpiAreaInfo.cs

@@ -2,5 +2,30 @@ namespace SJZY.Track.Domain.Dtos.Forewarning;
 
 public class KpiAreaInfo
 {
+    public int AreaId { get; set; }
     
+    /// <summary>
+    /// 国家或地区
+    /// </summary>
+    public string Country { get; set; } = null!;
+    
+    /// <summary>
+    /// 省份
+    /// </summary>
+    public string? Province { get; set; }
+    
+    /// <summary>
+    /// 城市
+    /// </summary>
+    public string? City { get; set; }
+    
+    /// <summary>
+    /// 开始邮编
+    /// </summary>
+    public string? LeftPostcode { get; set; }
+    
+    /// <summary>
+    /// 截止邮编
+    /// </summary>
+    public string? RightPostcode { get; set; }
 }

src/SJZY.Track.Domain/Dtos/Forewarning/WarningRuleCacheDto.cs

@@ -0,0 +1,20 @@
+namespace SJZY.Track.Domain.Dtos.Forewarning;
+
+public class WarningRuleCacheDto
+{
+    public int Id { get; set; }
+    public int SystemId { get; set; }
+    public int CustomerId { get; set; }
+    public int WarningId { get; set; }
+    public float TotalTimeSpan { get; set; }
+    public float WarnTimeSpan { get; set; }
+    public int WarnCount { get; set; }
+    public int StartTrackStatusId { get; set; }
+    public int EndTrackStatusId { get; set; }
+    public string? CargoType { get; set; }
+    public string? Months { get; set; }
+    public int? AreaId { get; set; }
+    public bool? IsTransit { get; set; }
+    public string? LeavePort { get; set; }
+    public string? ProductCode { get; set; }
+}

src/SJZY.Track.Repository/Repositories/KpiAreaRepository.cs

@@ -94,8 +94,18 @@ public class KpiAreaRepository : BasicRepository<RuleAreaEntity>, IKpiAreaReposi
     /// 获取所有区域信息
     /// </summary>
     /// <returns></returns>
-    public Task<List<RuleAreaInfoEntity>> GetAllAreaInfos()
+    public List<KpiAreaInfo> GetAllAreaInfos()
     {
-        return Context.Queryable<RuleAreaInfoEntity>().ToListAsync();
+        return Context.Queryable<RuleAreaInfoEntity>()
+            .Select((x)=>new KpiAreaInfo
+            {
+                AreaId = x.AreaId,
+                Country = x.CountryCode,
+                LeftPostcode = x.LeftPostcode,
+                City = x.City,
+                Province = x.Province,
+                RightPostcode = x.RightPostcode
+            })
+            .ToList();
     }
 }

src/SJZY.Track.Repository/Repositories/WarningRepository.cs

@@ -109,4 +109,30 @@ public class WarningRepository : BasicRepository<WarningEntity>, IWarningReposit
 
         return warning;
     }
+    
+    public Task<List<WarningRuleCacheDto>?> GetAllWarningRule()
+    {
+        return  Context.Queryable<WarningRuleEntity>()
+            .InnerJoin<WarningEntity>((a, b) => a.WarningId == b.Id)
+            .Where((a,b)=>b.IsEnable)
+            .Select((a,b)=>new WarningRuleCacheDto
+            {
+                Id = a.Id,
+                WarningId = a.WarningId,
+                TotalTimeSpan = a.TotalTimeSpan,
+                WarnTimeSpan = a.WarnTimeSpan,
+                WarnCount = a.WarnCount,
+                StartTrackStatusId = a.StartTrackStatusId,
+                EndTrackStatusId = a.EndTrackStatusId,
+                CargoType = a.CargoType,
+                Months = a.Months,
+                AreaId = a.AreaId,
+                IsTransit = a.IsTransit,
+                LeavePort = a.LeavePort,
+                ProductCode = a.ProductCode,
+                CustomerId = b.CustomerId,
+                SystemId = b.SubSystemId
+            })
+            .ToListAsync();
+    }
 }