TkEndpointCore.cs 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. using dodohold.core;
  2. using molilian.core;
  3. using Org.BouncyCastle.Crypto;
  4. using System;
  5. using System.Collections.Concurrent;
  6. using System.Linq;
  7. using YunhuiKit;
  8. using static molilian.core.TkPoolCore;
  9. public partial class TkEndpointCore
  10. {
  11. private static SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
  12. private static IEnumerable<TkEndpointConfigDTO> _cached;
  13. // 添加到 TkEndpointCore 类中
  14. private static readonly ConcurrentDictionary<(int, bool?), List<TkEndpointConfigDTO>> _accountEndpointCache = new();
  15. private static async Task<IEnumerable<TkEndpointConfigDTO>> ListAsync(bool force = false)
  16. {
  17. // 内存缓存检查
  18. if (!force && _cached != null) return _cached;
  19. string cache_key = "cache:tk_endpoint_config";
  20. if (!force)
  21. {
  22. var cachedList = await RedisHelper.GetAsync<IEnumerable<TkEndpointConfigDTO>>(cache_key);
  23. if (cachedList != null)
  24. {
  25. _cached = cachedList;
  26. return cachedList;
  27. }
  28. }
  29. // 获取新数据
  30. await _semaphore.WaitAsync();
  31. try
  32. {
  33. // 双重检查,防止并发情况下重复加载
  34. if (!force)
  35. {
  36. var cachedList = await RedisHelper.GetAsync<IEnumerable<TkEndpointConfigDTO>>(cache_key);
  37. if (cachedList != null)
  38. {
  39. _cached = cachedList;
  40. return cachedList;
  41. }
  42. }
  43. // 从数据库加载数据
  44. var list = new DBContext.Table("tk_endpoint_config")
  45. .Where("status=@status", new { status = 1 })
  46. .Select<TkEndpointConfigDTO>();
  47. if (list == null) return default;
  48. // 更新调用次数
  49. //foreach (var item in list)
  50. //{
  51. // item.current_daily_calls = await RiskControlCore.GetAllNodesTkEndpointCallsAsync(item.id, DateTime.Now.ToString("yyyyMMdd"));
  52. // item.current_hourly_calls = await RiskControlCore.GetAllNodesTkEndpointCallsAsync(item.id, DateTime.Now.ToString("yyyyMMddHH"));
  53. //}
  54. // 更新缓存
  55. await RedisHelper.SetAsync(cache_key, list, 30 * 86400);
  56. _cached = list;
  57. _accountEndpointCache.Clear();
  58. return list;
  59. }
  60. finally
  61. {
  62. _semaphore.Release();
  63. }
  64. }
  65. public static async Task<List<TkEndpointConfigDTO>> GetEndpointsByAccountAsync(int accountId, bool? isTaobaoUrl = null, string parseEndpoints = null)
  66. {
  67. var cacheKey = (accountId, isTaobaoUrl);
  68. if (_accountEndpointCache.TryGetValue(cacheKey, out var cached))
  69. return cached.Select(x => x.DeepCopy()).ToList();
  70. // 2. 获取所有端点(包含全局和专属)
  71. var allEndpoints = (await ListAsync())?.Where(e => e.status).ToList() ?? [];
  72. // 3. 合并逻辑:专属配置覆盖全局配置(关键!)
  73. var mergedEndpoints = allEndpoints
  74. .GroupBy(e => e.endpoint)
  75. .Select(g =>
  76. {
  77. // 优先取专属配置
  78. var accountSpecific = g.FirstOrDefault(e => e.tk_pool_id == accountId);
  79. if (accountSpecific != null)
  80. return accountSpecific.DeepCopy(); // 深拷贝专属配置
  81. // 否则取全局配置(tk_pool_id=0)
  82. var global = g.FirstOrDefault(e => e.tk_pool_id == 0);
  83. return global?.DeepCopy(); // 深拷贝全局配置
  84. })
  85. .Where(e => e != null)
  86. .ToList();
  87. // 4. 根据isTaobaoUrl过滤
  88. if (isTaobaoUrl.HasValue)
  89. {
  90. mergedEndpoints = mergedEndpoints.Where(e =>
  91. {
  92. if (e.taobao_id_handling == 0) return true;
  93. if (e.taobao_id_handling == 1 && isTaobaoUrl.Value) return true;
  94. if (e.taobao_id_handling == 2 && !isTaobaoUrl.Value) return true;
  95. return false;
  96. }).ToList();
  97. }
  98. // 5. 排除指定的端点
  99. if (!string.IsNullOrEmpty(parseEndpoints))
  100. {
  101. var parseEndpointIds = parseEndpoints.Split(',')
  102. .Select(idStr => int.TryParse(idStr.Trim(), out var id) ? id : (int?)null)
  103. .Where(id => id.HasValue)
  104. .Select(id => id.Value)
  105. .ToHashSet();
  106. mergedEndpoints = mergedEndpoints.Where(e => parseEndpointIds.Contains(e.ep_id)).ToList();
  107. }
  108. // 6. 存入缓存
  109. _accountEndpointCache.TryAdd(cacheKey, mergedEndpoints.Select(x => x.DeepCopy()).ToList());
  110. return mergedEndpoints;
  111. }
  112. // 添加缓存清理方法
  113. public static void ClearEndpointCache(int accountId)
  114. {
  115. _accountEndpointCache.TryRemove((accountId, true), out _);
  116. _accountEndpointCache.TryRemove((accountId, false), out _);
  117. _accountEndpointCache.TryRemove((accountId, null), out _);
  118. }
  119. public static void Refresh()
  120. {
  121. _ = AllListAsync(true);
  122. _ = ListAsync(true);
  123. _accountEndpointCache.Clear();
  124. }
  125. }