src/JiShe.CollectBus.Application.Contracts/RedisDataCache/IRedisDataCacheService.cs

@@ -1,187 +0,0 @@
-using JiShe.CollectBus.Common.Models;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Application.Contracts
-{
-    /// <summary>
-    /// 数据缓存服务接口
-    /// </summary>
-    public interface IRedisDataCacheService
-    {
-        /// <summary>
-        /// 单个添加数据
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="data">待缓存数据</param>
-        /// <returns></returns>
-        Task InsertDataAsync<T>(
-        string redisHashCacheKey,
-        string redisSetIndexCacheKey,
-        string redisZSetScoresIndexCacheKey,
-        T data) where T : DeviceCacheBasicModel;
-
-        /// <summary>
-        /// 批量添加数据
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="items">待缓存数据集合</param>
-        /// <returns></returns>
-        Task BatchInsertDataAsync<T>(
-        string redisHashCacheKey,
-        string redisSetIndexCacheKey,
-        string redisZSetScoresIndexCacheKey,
-        IEnumerable<T> items) where T : DeviceCacheBasicModel;
-
-        /// <summary>
-        /// 删除缓存信息
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="data">已缓存数据</param>
-        /// <returns></returns>
-        Task RemoveCacheDataAsync<T>(
-        string redisHashCacheKey,
-        string redisSetIndexCacheKey,
-        string redisZSetScoresIndexCacheKey,
-        T data) where T : DeviceCacheBasicModel;
-
-        /// <summary>
-        /// 修改缓存信息,映射关系未发生改变
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="newData">待修改缓存数据</param>
-        /// <returns></returns>
-        Task ModifyDataAsync<T>(
-        string redisHashCacheKey,
-        string redisSetIndexCacheKey,
-        string redisZSetScoresIndexCacheKey,
-        T newData) where T : DeviceCacheBasicModel;
-
-
-        /// <summary>
-        /// 修改缓存信息,映射关系已改变
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisSetIndexCacheKey">Set索引缓存Key</param>
-        /// <param name="oldMemberId">旧的映射关系</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="newData">待修改缓存数据</param>
-        /// <returns></returns>
-        Task ModifyDataAsync<T>(
-        string redisHashCacheKey,
-        string redisSetIndexCacheKey,
-        string oldMemberId,
-        string redisZSetScoresIndexCacheKey,
-        T newData) where T : DeviceCacheBasicModel;
-
-        ///// <summary>
-        ///// 通过集中器与表计信息排序索引获取数据
-        ///// </summary>
-        ///// <typeparam name="T"></typeparam>
-        ///// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        ///// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        ///// <param name="pageSize">分页尺寸</param>
-        ///// <param name="lastScore">最后一个索引</param>
-        ///// <param name="lastMember">最后一个唯一标识</param>
-        ///// <param name="descending">排序方式</param>
-        ///// <returns></returns>
-        //Task<BusCacheGlobalPagedResult<T>> GetPagedData<T>(
-        //string redisHashCacheKey,
-        //string redisZSetScoresIndexCacheKey,
-        //IEnumerable<int> focusIds,
-        //int pageSize = 10,
-        //decimal? lastScore = null,
-        //string lastMember = null,
-        //bool descending = true)
-        //where T : DeviceCacheBasicModel;
-
-
-        /// <summary>
-        /// 通过ZSET索引获取数据，支持10万级别数据处理，控制在13秒以内。
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="redisHashCacheKey">主数据存储Hash缓存Key</param>
-        /// <param name="redisZSetScoresIndexCacheKey">ZSET索引缓存Key</param>
-        /// <param name="pageSize">分页尺寸</param>
-        /// <param name="lastScore">最后一个索引</param>
-        /// <param name="lastMember">最后一个唯一标识</param>
-        /// <param name="descending">排序方式</param>
-        /// <returns></returns>
-        Task<BusCacheGlobalPagedResult<T>> GetAllPagedData<T>(
-        string redisHashCacheKey,
-        string redisZSetScoresIndexCacheKey,
-        int pageSize = 1000,
-        decimal? lastScore = null,
-        string lastMember = null,
-        bool descending = true)
-        where T : DeviceCacheBasicModel;
-
-
-        /// <summary>
-        /// 优化后的分页获取方法（支持百万级数据）
-        /// </summary>
-        Task<BusCacheGlobalPagedResult<T>> GetAllPagedDataOptimized<T>(
-            string redisHashCacheKey,
-            string redisZSetScoresIndexCacheKey,
-            int pageSize = 1000,
-            decimal? lastScore = null,
-            string lastMember = null,
-            bool descending = true) where T : DeviceCacheBasicModel;
-          
-        ///// <summary>
-        ///// 游标分页查询
-        ///// </summary>
-        ///// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
-        ///// <param name="pageSize">分页数量</param>
-        ///// <param name="startScore">开始索引</param>
-        ///// <param name="excludeMember">开始唯一标识</param>
-        ///// <param name="descending">排序方式</param>
-        ///// <returns></returns>
-        //Task<(List<string> Members, bool HasNext)> GetPagedMembers(
-        //    string redisZSetScoresIndexCacheKey,
-        //    int pageSize,
-        //    decimal? startScore,
-        //    string excludeMember,
-        //    bool descending);
-
-            ///// <summary>
-            ///// 批量获取指定分页的数据
-            ///// </summary>
-            ///// <typeparam name="T"></typeparam>
-            ///// <param name="redisHashCacheKey">Hash表缓存key</param>
-            ///// <param name="members">Hash表字段集合</param>
-            ///// <returns></returns>
-            //Task<Dictionary<string, T>> BatchGetData<T>(
-            //    string redisHashCacheKey,
-            //    IEnumerable<string> members)
-            //    where T : DeviceCacheBasicModel;
-
-            ///// <summary>
-            ///// 获取下一页游标
-            ///// </summary>
-            ///// <param name="redisZSetScoresIndexCacheKey">排序索引ZSET缓存Key</param>
-            ///// <param name="lastMember">最后一个唯一标识</param>
-            ///// <param name="descending">排序方式</param>
-            ///// <returns></returns>
-            //Task<decimal?> GetNextScore(
-            //    string redisZSetScoresIndexCacheKey,
-            //    string lastMember,
-            //    bool descending);
-    }
-}

src/JiShe.CollectBus.Application/CollectBusAppService.cs

@@ -79,7 +79,7 @@ public abstract class CollectBusAppService : ApplicationService
                 {
                     string key = (string)item[0];
                     object[] fieldsAndValues = (object[])item[1];
-                    var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, systemType, serverTagName, meterType, timeDensity)}";
+                    var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, systemType, serverTagName, meterType, timeDensity)}";
                     string focusAddress = key.Replace(redisCacheKey, "");
 
                     var meterHashs = new Dictionary<string, T>();
@@ -182,7 +182,7 @@ public abstract class CollectBusAppService : ApplicationService
                     string key = (string)item[0];
                     object[] fieldsAndValues = (object[])item[1];
                     var redisCacheKey = string.Format(
-                        RedisConst.CacheMeterInfoHashKey,
+                        RedisConst.CacheMeterInfoKey,
                         systemType,
                         serverTagName,
                         meterType,

src/JiShe.CollectBus.Application/Samples/SampleAppService.cs

@@ -23,9 +23,6 @@ using JiShe.CollectBus.Common.DeviceBalanceControl;
 using JiShe.CollectBus.Kafka.Attributes;
 using System.Text.Json;
 using JiShe.CollectBus.Kafka;
-using JiShe.CollectBus.Application.Contracts;
-using JiShe.CollectBus.Common.Models;
-using System.Diagnostics;
 
 namespace JiShe.CollectBus.Samples;
 
@@ -35,23 +32,17 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
     private readonly IIoTDBProvider _iotDBProvider;
     private readonly IoTDBRuntimeContext _dbContext;
     private readonly IoTDBOptions _options;
-    private readonly IRedisDataCacheService _redisDataCacheService;
 
     public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options,
-        IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService)
+        IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger)
     {
         _iotDBProvider = iotDBProvider;
         _options = options.Value;
         _dbContext = dbContext;
         _logger = logger;
-        _redisDataCacheService = redisDataCacheService;
     }
 
-    /// <summary>
+
-    /// 测试 UseSessionPool
-    /// </summary>
-    /// <param name="timestamps"></param>
-    /// <returns></returns>
     [HttpGet]
     public async Task UseSessionPool(long timestamps)
     {
@@ -81,10 +72,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
         await _iotDBProvider.InsertAsync(meter);
     }
 
-    /// <summary>
+
-    /// 测试Session切换
-    /// </summary>
-    /// <returns></returns>
     [HttpGet]
     public async Task UseTableSessionPool()
     {
@@ -137,7 +125,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
 
         var timeDensity = "15";
         //获取缓存中的电表信息
-        var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";
+        var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";
 
         var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
         var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, "Energy", "JiSheCollectBus", timeDensity, MeterTypeEnum.Ammeter);
@@ -190,44 +178,6 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
         await _iotDBProvider.InsertAsync(meter);
     }
 
-    /// <summary>
-    /// 测试Redis批量读取10万条数据性能
-    /// </summary>
-    /// <returns></returns>
-    [HttpGet]
-    public async Task TestRedisCacheGetAllPagedData()
-    {
-        var timeDensity = "15";
-        string SystemType = "Energy";
-        string ServerTagName = "JiSheCollectBus2";
-        var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-        var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-        var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-
-        var timer1 = Stopwatch.StartNew();
-        decimal? cursor = null;
-        string member = null;
-        bool hasNext;
-        List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
-        do
-        {
-            var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
-                redisCacheMeterInfoHashKeyTemp,
-                redisCacheMeterInfoZSetScoresIndexKeyTemp,
-                pageSize: 1000,
-                lastScore: cursor,
-                lastMember: member);
-
-            meterInfos.AddRange(page.Items);
-            cursor = page.HasNext ? page.NextScore : null;
-            member = page.HasNext ? page.NextMember : null;
-            hasNext = page.HasNext;
-        } while (hasNext);
-
-        timer1.Stop();
-        _logger.LogError($"读取数据更花费时间{timer1.ElapsedMilliseconds}毫秒");
-    }
-
 
     public Task<SampleDto> GetAsync()
     {

src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs

@@ -1,13 +1,11 @@
 using DotNetCore.CAP;
 using JiShe.CollectBus.Ammeters;
-using JiShe.CollectBus.Application.Contracts;
 using JiShe.CollectBus.Common.BuildSendDatas;
 using JiShe.CollectBus.Common.Consts;
 using JiShe.CollectBus.Common.DeviceBalanceControl;
 using JiShe.CollectBus.Common.Enums;
 using JiShe.CollectBus.Common.Extensions;
 using JiShe.CollectBus.Common.Helpers;
-using JiShe.CollectBus.Common.Models;
 using JiShe.CollectBus.GatherItem;
 using JiShe.CollectBus.IoTDBProvider;
 using JiShe.CollectBus.IotSystems.MessageIssueds;
@@ -15,16 +13,13 @@ using JiShe.CollectBus.IotSystems.MeterReadingRecords;
 using JiShe.CollectBus.IotSystems.Watermeter;
 using JiShe.CollectBus.Kafka.Producer;
 using JiShe.CollectBus.Protocol.Contracts;
-using JiShe.CollectBus.RedisDataCache;
 using JiShe.CollectBus.Repository.MeterReadingRecord;
-using Mapster;
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
-using static FreeSql.Internal.GlobalFilter;
 
 namespace JiShe.CollectBus.ScheduledMeterReading
 {
@@ -37,7 +32,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
         private readonly IIoTDBProvider _dbProvider;
         private readonly IMeterReadingRecordRepository _meterReadingRecordRepository;
         private readonly IProducerService _producerService;
-        private readonly IRedisDataCacheService _redisDataCacheService;
         private readonly ICapPublisher _producerBus;
 
         public BasicScheduledMeterReadingService(
@@ -45,7 +39,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             ICapPublisher producerBus,
             IMeterReadingRecordRepository meterReadingRecordRepository,
             IProducerService producerService,
-            IRedisDataCacheService redisDataCacheService,
             IIoTDBProvider dbProvider)
         {
             _producerBus = producerBus;
@@ -53,7 +46,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             _dbProvider = dbProvider;
             _meterReadingRecordRepository = meterReadingRecordRepository;
             _producerService = producerService;
-            _redisDataCacheService = redisDataCacheService;
         }
 
         /// <summary>
@@ -124,44 +116,31 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                     continue;
                 }
 
+
+
+                //获取缓存中的表信息
+                var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, meteryType, timeDensity)}*";
+                var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
+                if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
+                {
+                    _logger.LogError($"{nameof(CreateToBeIssueTasks)} {timeDensity}分钟采集待下发任务创建失败，没有获取到缓存信息,-104");
+                    return;
+                }
+
                 var meterTypes = EnumExtensions.ToEnumDictionary<MeterTypeEnum>();
 
                 if (meteryType == MeterTypeEnum.Ammeter.ToString())
                 {
-                    var timer = Stopwatch.StartNew();
+                    // 解析结果（结果为嵌套数组）
-
+                    var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, SystemType, ServerTagName, $"{timeDensity}", meterTypes[meteryType]);
-                    //获取对应频率中的所有电表信息
-                    var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-                    var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-                    var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
-
-                    List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
-                    decimal? cursor = null;
-                    string member = null;
-                    bool hasNext;
-                    do
-                    {
-                        var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
-                            redisCacheMeterInfoHashKeyTemp,
-                            redisCacheMeterInfoZSetScoresIndexKeyTemp,
-                            pageSize: 1000,
-                            lastScore: cursor,
-                            lastMember: member);
-
-                        meterInfos.AddRange(page.Items);
-                        cursor = page.HasNext ? page.NextScore : null;
-                        member = page.HasNext ? page.NextMember : null;
-                        hasNext = page.HasNext;
-                    } while (hasNext);
-                     
                     if (meterInfos == null || meterInfos.Count <= 0)
                     {
-                        timer.Stop();
                         _logger.LogError($"{nameof(CreateToBeIssueTasks)} {timeDensity}分钟采集待下发任务创建失败，没有获取到缓存信息,-105");
                         return;
                     }
                     //await AmmerterScheduledMeterReadingIssued(timeDensity, meterInfos);
 
+                    var timer = Stopwatch.StartNew();
 
                     //处理数据 
                     //await DeviceGroupBalanceControl.ProcessGenericListAsync(
@@ -178,14 +157,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                     await DeviceGroupBalanceControl.ProcessWithThrottleAsync(
                         items: meterInfos,
                         deviceIdSelector: data => data.FocusAddress,
-                        processor: (data,groupIndex) =>
+                        processor: data =>
                         {
-                            _ = AmmerterCreatePublishTask(timeDensity, data, groupIndex,tasksToBeIssueModel.NextTaskTime.ToString("yyyyMMddHHmmss"));
+                            _ = AmmerterCreatePublishTask(timeDensity, data);
                         }
                     );
 
                     timer.Stop();
-                    _logger.LogInformation($"{nameof(CreateToBeIssueTasks)} {timeDensity}分钟采集待下发任务创建完成,{timer.ElapsedMilliseconds},总共{meterInfos.Count}表计信息");
+                    _logger.LogInformation($"{nameof(CreateToBeIssueTasks)} {timeDensity}分钟采集待下发任务创建完成,{timer.ElapsedMilliseconds},{oneMinutekeyList.Length}");
 
                 }
                 else if (meteryType == MeterTypeEnum.WaterMeter.ToString())
@@ -228,86 +207,25 @@ namespace JiShe.CollectBus.ScheduledMeterReading
         public virtual async Task InitAmmeterCacheData(string gatherCode = "")
         {
 #if DEBUG
-            //var timeDensity = "15";
-            //string tempCacheMeterInfoKey = $"CollectBus:{"{0}:{1}"}:MeterInfo:{"{2}"}:{"{3}"}";
-            ////获取缓存中的电表信息
-            //var redisKeyList = $"{string.Format(tempCacheMeterInfoKey, SystemType, "JiSheCollectBus", MeterTypeEnum.Ammeter, timeDensity)}*";
-
-            //var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
-            //var tempMeterInfos = await GetMeterRedisCacheListData<AmmeterInfoTemp>(oneMinutekeyList, SystemType, ServerTagName, timeDensity, MeterTypeEnum.Ammeter);
-            ////List<string> focusAddressDataLista = new List<string>();
-            //List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
-            //foreach (var item in tempMeterInfos)
-            //{
-            //    var tempData = item.Adapt<AmmeterInfo>();
-            //    tempData.FocusId = item.FocusID;
-            //    tempData.MeterId = item.Id;
-            //    meterInfos.Add(tempData);
-            //    //focusAddressDataLista.Add(item.FocusAddress);
-            //}
-            
-            
-            
             var timeDensity = "15";
-            var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
+            //获取缓存中的电表信息
-            var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
+            var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}*";
-            var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
 
-            List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
+            var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
+            var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, SystemType, ServerTagName, timeDensity, MeterTypeEnum.Ammeter);
             List<string> focusAddressDataLista = new List<string>();
-            var timer1 = Stopwatch.StartNew();
+            foreach (var item in meterInfos)
-            //decimal? cursor = null;
-            //string member = null;
-            //bool hasNext;
-            //do
-            //{
-            //    var page = await _redisDataCacheService.GetAllPagedDataOptimized<AmmeterInfo>(
-            //        redisCacheMeterInfoHashKeyTemp,
-            //        redisCacheMeterInfoZSetScoresIndexKeyTemp,
-            //        pageSize: 1000,
-            //        lastScore: cursor,
-            //        lastMember: member);
-
-            //    meterInfos.AddRange(page.Items);
-            //    cursor = page.HasNext ? page.NextScore : null;
-            //    member = page.HasNext ? page.NextMember : null;
-            //    hasNext = page.HasNext;
-            //} while (hasNext);
-
-            var allIds = new HashSet<string>();
-            decimal? score = null;
-            string member = null;
-
-            while (true)
             {
-                var page = await _redisDataCacheService.GetAllPagedDataOptimized<AmmeterInfo>(
+                focusAddressDataLista.Add(item.FocusAddress);
-                    redisCacheMeterInfoHashKeyTemp,
-                    redisCacheMeterInfoZSetScoresIndexKeyTemp,
-                    pageSize: 1000,
-                    lastScore: score,
-                    lastMember: member);
-
-                meterInfos.AddRange(page.Items);
-                focusAddressDataLista.AddRange(page.Items.Select(d=>d.FocusAddress));
-                foreach (var item in page.Items)
-                {
-                    if (!allIds.Add(item.MemberId))
-                        throw new Exception("Duplicate data found!");
-                }
-                if (!page.HasNext) break;
-                score = page.NextScore;
-                member = page.NextMember;
             }
 
-
+            DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);
-            timer1.Stop();
+            return;
-            _logger.LogError($"读取数据更花费时间{timer1.ElapsedMilliseconds}毫秒");
-            //DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);
-            //return;
 #else
             var meterInfos = await GetAmmeterInfoList(gatherCode);
 #endif
 
+
             if (meterInfos == null || meterInfos.Count <= 0)
             {
                 throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，电表数据为空");
@@ -319,7 +237,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             {
                 throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，采集项类型数据为空");
             }
-            var timer = Stopwatch.StartNew();
 
             List<string> focusAddressDataList = new List<string>();//用于处理Kafka主题分区数据的分发和处理。
 
@@ -327,11 +244,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             var meterInfoGroupByTimeDensity = meterInfos.GroupBy(d => d.TimeDensity);
             foreach (var itemTimeDensity in meterInfoGroupByTimeDensity)
             {
-                var redisCacheMeterInfoHashKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
-                var redisCacheMeterInfoSetIndexKey = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}";
-                var redisCacheMeterInfoZSetScoresIndexKey = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}"; 
-
-                List<AmmeterInfo> ammeterInfos = new List<AmmeterInfo>();
                 //将表计信息根据集中器分组，获得集中器号
                 var meterInfoGroup = itemTimeDensity.GroupBy(x => x.FocusAddress).ToList();
                 foreach (var item in meterInfoGroup)
@@ -343,17 +255,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
                     focusAddressDataList.Add(item.Key);
 
-                   // var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}{item.Key}";
+                    var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, itemTimeDensity.Key)}{item.Key}";
 
 #if DEBUG
                     //每次缓存时，删除缓存，避免缓存数据有不准确的问题
                     //await FreeRedisProvider.Instance.DelAsync(redisCacheKey);
 #else
                    //每次缓存时，删除缓存，避免缓存数据有不准确的问题
-                   //await FreeRedisProvider.Instance.DelAsync(redisCacheKey);
+                   await FreeRedisProvider.Instance.DelAsync(redisCacheKey);
 #endif
 
-                    //Dictionary<string, AmmeterInfo> keyValuePairs = new Dictionary<string, AmmeterInfo>();
+                    Dictionary<string, AmmeterInfo> keyValuePairs = new Dictionary<string, AmmeterInfo>();
                     foreach (var ammeter in item)
                     {
                         //处理ItemCode
@@ -398,17 +310,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                             }
                         }
 
-                        ammeterInfos.Add(ammeter);
+                        keyValuePairs.TryAdd($"{ammeter.ID}", ammeter);
-                        //keyValuePairs.TryAdd($"{ammeter.MeterId}", ammeter);
                     }
-                    //await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
+                    await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
                 }
 
-                await _redisDataCacheService.BatchInsertDataAsync<AmmeterInfo>(
-                    redisCacheMeterInfoHashKey,
-                    redisCacheMeterInfoSetIndexKey,
-                    redisCacheMeterInfoZSetScoresIndexKey,ammeterInfos);
-
                 //在缓存表信息数据的时候，新增下一个时间的自动处理任务，1分钟后执行所有的采集频率任务
                 TasksToBeIssueModel nextTask = new TasksToBeIssueModel()
                 {
@@ -431,9 +337,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 DeviceGroupBalanceControl.InitializeCache(focusAddressDataList);
             }
 
-            timer.Stop();
+            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成");
-
-            _logger.LogInformation($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据完成，耗时{timer.ElapsedMilliseconds}毫秒");
         }
 
         /// <summary>
@@ -704,11 +608,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
         /// </summary>
         /// <param name="timeDensity">采集频率</param>
         /// <param name="ammeterInfo">集中器号hash分组的集中器集合数据</param>
-        /// <param name="groupIndex">集中器所在分组</param>
-        /// <param name="taskBatch">时间格式的任务批次名称</param>
         /// <returns></returns>
         private async Task AmmerterCreatePublishTask(int timeDensity
-            , AmmeterInfo ammeterInfo,int groupIndex,string taskBatch)
+            , AmmeterInfo ammeterInfo)
         {
             var handlerPacketBuilder = TelemetryPacketBuilder.AFNHandlersDictionary;
             //todo 检查需要待补抄的电表的时间点信息，保存到需要待补抄的缓存中。如果此线程异常，该如何补偿？
@@ -716,9 +618,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             var currentTime = DateTime.Now;
             var pendingCopyReadTime = currentTime.AddMinutes(timeDensity);
             //构建缓存任务key，依然 表计类型+采集频率+集中器地址，存hash类型
-            var redisCacheTelemetryPacketInfoHashKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity, groupIndex, taskBatch)}";
+            var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{ammeterInfo.FocusAddress}";
-            var redisCacheTelemetryPacketInfoSetIndexKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity, groupIndex, taskBatch)}";
-            var redisCacheTelemetryPacketInfoZSetScoresIndexKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity, groupIndex, taskBatch)}";
 
             if (string.IsNullOrWhiteSpace(ammeterInfo.ItemCodes))
             {
@@ -799,8 +699,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                 }
             }
 
-            //Dictionary<string, MeterReadingRecords> keyValuePairs = new Dictionary<string, MeterReadingRecords>();
+            Dictionary<string, MeterReadingRecords> keyValuePairs = new Dictionary<string, MeterReadingRecords>();
-            List<MeterReadingTelemetryPacketInfo> taskList = new List<MeterReadingTelemetryPacketInfo>();
 
             foreach (var tempItem in tempCodes)
             {
@@ -855,17 +754,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
 
 
-                var meterReadingRecords = new MeterReadingTelemetryPacketInfo()
+                var meterReadingRecords = new MeterReadingRecords()
                 {
                     ProjectID = ammeterInfo.ProjectID,
                     DatabaseBusiID = ammeterInfo.DatabaseBusiID,
                     PendingCopyReadTime = pendingCopyReadTime,
                     CreationTime = currentTime,
                     MeterAddress = ammeterInfo.AmmerterAddress,
-                    MeterId = ammeterInfo.MeterId,
+                    MeterId = ammeterInfo.ID,
                     MeterType = MeterTypeEnum.Ammeter,
                     FocusAddress = ammeterInfo.FocusAddress,
-                    FocusId = ammeterInfo.FocusId,
+                    FocusID = ammeterInfo.FocusID,
                     AFN = aFN,
                     Fn = fn,
                     ItemCode = tempItem,
@@ -875,10 +774,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                     IssuedMessageId = GuidGenerator.Create().ToString(),
                     IssuedMessageHexString = Convert.ToHexString(dataInfos),
                 };
+                meterReadingRecords.CreateDataId(GuidGenerator.Create());
 
-                //meterReadingRecords.CreateDataId(GuidGenerator.Create());
+                keyValuePairs.TryAdd($"{ammeterInfo.ID}_{tempItem}", meterReadingRecords);
-
-                taskList.Add(meterReadingRecords);
             }
             //TimeSpan timeSpan = TimeSpan.FromMicroseconds(5);
             //await Task.Delay(timeSpan);
@@ -886,25 +784,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             //return keyValuePairs;
             // await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
 
-            //using (var pipe = FreeRedisProvider.Instance.StartPipe())
+            using (var pipe = FreeRedisProvider.Instance.StartPipe())
-            //{
-            //    pipe.HSet(redisCacheKey, keyValuePairs);
-            //    object[] ret = pipe.EndPipe();
-            //}
-            if (taskList == null
-              || taskList.Count() <= 0
-              || string.IsNullOrWhiteSpace(redisCacheTelemetryPacketInfoHashKey)
-              || string.IsNullOrWhiteSpace(redisCacheTelemetryPacketInfoSetIndexKey)
-              || string.IsNullOrWhiteSpace(redisCacheTelemetryPacketInfoZSetScoresIndexKey))
             {
-                _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 写入参数异常，{redisCacheTelemetryPacketInfoHashKey}：{redisCacheTelemetryPacketInfoSetIndexKey}：{redisCacheTelemetryPacketInfoZSetScoresIndexKey}，-101");
+                pipe.HSet(redisCacheKey, keyValuePairs);
-                return;
+                object[] ret = pipe.EndPipe();
             }
-            await _redisDataCacheService.BatchInsertDataAsync(
+
-                redisCacheTelemetryPacketInfoHashKey,
+             
-                redisCacheTelemetryPacketInfoSetIndexKey,
+            await Task.CompletedTask;
-                redisCacheTelemetryPacketInfoZSetScoresIndexKey,
-                taskList);
         }
 
         /// <summary>
@@ -994,7 +881,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             foreach (var focusInfo in focusGroup)
             {
                 //构建缓存任务key，依然 表计类型+采集频率+集中器地址，存hash类型
-                var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{focusInfo.Key}";
+                var redisCacheKey = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}{focusInfo.Key}";
 
                 foreach (var ammeterInfo in focusInfo.Value)
                 {
@@ -1028,22 +915,22 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
                     if (string.IsNullOrWhiteSpace(ammeter.AreaCode))
                     {
-                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.MeterId},集中器通信区号为空");
+                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.ID},集中器通信区号为空");
                         continue;
                     }
                     if (string.IsNullOrWhiteSpace(ammeter.Address))
                     {
-                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.MeterId},集中器通信地址为空");
+                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.ID},集中器通信地址为空");
                         continue;
                     }
                     if (Convert.ToInt32(ammeter.Address) > 65535)
                     {
-                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.MeterId},集中器通信地址无效,确保大于65535");
+                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.ID},集中器通信地址无效,确保大于65535");
                         continue;
                     }
                     if (ammeter.MeteringCode <= 0 || ammeter.MeteringCode > 2033)
                     {
-                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.MeterId},非有效测量点号({ammeter.MeteringCode})");
+                        _logger.LogError($"{nameof(AmmerterCreatePublishTask)} 表ID:{ammeter.ID},非有效测量点号({ammeter.MeteringCode})");
                         continue;
                     }
 
@@ -1141,10 +1028,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                             PendingCopyReadTime = pendingCopyReadTime,
                             CreationTime = currentTime,
                             MeterAddress = ammeter.AmmerterAddress,
-                            MeterId = ammeter.MeterId,
+                            MeterId = ammeter.ID,
                             MeterType = MeterTypeEnum.Ammeter,
                             FocusAddress = ammeter.FocusAddress,
-                            FocusID = ammeter.FocusId,
+                            FocusID = ammeter.FocusID,
                             AFN = aFN,
                             Fn = fn,
                             ItemCode = tempItem,
@@ -1154,9 +1041,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                             IssuedMessageId = GuidGenerator.Create().ToString(),
                             IssuedMessageHexString = Convert.ToHexString(dataInfos),
                         };
-                        //meterReadingRecords.CreateDataId(GuidGenerator.Create());
+                        meterReadingRecords.CreateDataId(GuidGenerator.Create());
 
-                        keyValuePairs.TryAdd($"{ammeter.MeterId}_{tempItem}", meterReadingRecords);
+                        keyValuePairs.TryAdd($"{ammeter.ID}_{tempItem}", meterReadingRecords);
                     }
                     await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
                 }
@@ -1210,12 +1097,12 @@ namespace JiShe.CollectBus.ScheduledMeterReading
                         continue;
                     }
 
-                    var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.WaterMeter, itemTimeDensity.Key)}{item.Key}";
+                    var redisCacheKey = $"{string.Format(RedisConst.CacheMeterInfoKey, SystemType, ServerTagName, MeterTypeEnum.WaterMeter, itemTimeDensity.Key)}{item.Key}";
                     Dictionary<string, WatermeterInfo> keyValuePairs = new Dictionary<string, WatermeterInfo>();
                     foreach (var subItem in item)
                     {
 
-                        keyValuePairs.TryAdd($"{subItem.MeterId}", subItem);
+                        keyValuePairs.TryAdd($"{subItem.ID}", subItem);
                     }
                     await FreeRedisProvider.Instance.HSetAsync(redisCacheKey, keyValuePairs);
                 }
@@ -1358,7 +1245,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
         /// <returns></returns>
         private string GetTelemetryPacketCacheKeyPrefix(int timeDensity, MeterTypeEnum meterType)
         {
-            return $"{string.Format(RedisConst.CacheTelemetryPacketInfoHashKey, SystemType, ServerTagName, meterType, timeDensity)}*";
+            return $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, ServerTagName, meterType, timeDensity)}*";
         }
 
         #endregion

src/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs

@@ -4,7 +4,6 @@ using System.Threading.Tasks;
 using Confluent.Kafka;
 using DotNetCore.CAP;
 using JiShe.CollectBus.Ammeters;
-using JiShe.CollectBus.Application.Contracts;
 using JiShe.CollectBus.Common.Consts;
 using JiShe.CollectBus.Common.DeviceBalanceControl;
 using JiShe.CollectBus.Common.Helpers;
@@ -36,19 +35,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
     public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
     {
         string serverTagName = string.Empty;
-        public EnergySystemScheduledMeterReadingService(
+        public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger,
-            ILogger<EnergySystemScheduledMeterReadingService> logger,
+         ICapPublisher producerBus, IIoTDBProvider dbProvider, IMeterReadingRecordRepository meterReadingRecordRepository,IConfiguration configuration, IProducerService  producerService) : base(logger, producerBus, meterReadingRecordRepository, producerService,dbProvider)
-            ICapPublisher producerBus, IIoTDBProvider dbProvider,
-            IMeterReadingRecordRepository meterReadingRecordRepository,
-            IConfiguration configuration, 
-            IProducerService producerService, 
-            IRedisDataCacheService redisDataCacheService) 
-            : base(logger,
-                  producerBus, 
-                  meterReadingRecordRepository,
-                  producerService,
-                  redisDataCacheService, 
-                  dbProvider)
         {
             serverTagName = configuration.GetValue<string>(CommonConst.ServerTagName)!;
         }
@@ -92,11 +80,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             //    Baudrate = 2400,
             //    FocusAddress = "402440506",
             //    Name = "张家祠工务(三相电表)",
-            //    FocusId = 95780,
+            //    FocusID = 95780,
             //    DatabaseBusiID = 1,
             //    MeteringCode = 1,
             //    AmmerterAddress = "402410040506",
-            //    MeterId = 127035,
+            //    ID = 127035,
             //    TypeName = 3,
             //    DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
             //    TimeDensity = 15,
@@ -106,11 +94,11 @@ namespace JiShe.CollectBus.ScheduledMeterReading
             //    Baudrate = 2400,
             //    FocusAddress = "542400504",
             //    Name = "五号配（长芦二所四排）(单相电表)",
-            //    FocusId = 69280,
+            //    FocusID = 69280,
             //    DatabaseBusiID = 1,
             //    MeteringCode = 2,
             //    AmmerterAddress = "542410000504",
-            //    MeterId = 95594,
+            //    ID = 95594,
             //    TypeName = 1,
             //    DataTypes = "581,589,592,597,601",
             //    TimeDensity = 15,
@@ -118,7 +106,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
 
             //return ammeterInfos;
 
-            string sql = $@"SELECT C.ID as MeterId,C.Name,C.FocusID as FocusId,C.SingleRate,C.MeteringCode,C.Code AS BrandType,C.Baudrate,C.Password,C.MeteringPort,C.[Address] AS AmmerterAddress,C.TypeName,C.Protocol,C.TripState,C.[State],B.[Address],B.AreaCode,B.AutomaticReport,D.DataTypes,B.TimeDensity,A.GatherCode,C.Special,C.[ProjectID],B.AbnormalState,B.LastTime,CONCAT(B.AreaCode, B.[Address]) AS FocusAddress,(select top 1 DatabaseBusiID from TB_Project where ID = B.ProjectID) AS DatabaseBusiID
+            string sql = $@"SELECT C.ID,C.Name,C.FocusID,C.SingleRate,C.MeteringCode,C.Code AS BrandType,C.Baudrate,C.Password,C.MeteringPort,C.[Address] AS AmmerterAddress,C.TypeName,C.Protocol,C.TripState,C.[State],B.[Address],B.AreaCode,B.AutomaticReport,D.DataTypes,B.TimeDensity,A.GatherCode,C.Special,C.[ProjectID],B.AbnormalState,B.LastTime,CONCAT(B.AreaCode, B.[Address]) AS FocusAddress,(select top 1 DatabaseBusiID from TB_Project where ID = B.ProjectID) AS DatabaseBusiID
               FROM TB_GatherInfo(NOLOCK) AS A
               INNER JOIN TB_FocusInfo(NOLOCK) AS B ON A.ID = B.GatherInfoID AND B.RemoveState >= 0 AND B.State>=0
               INNER JOIN TB_AmmeterInfo(NOLOCK) AS C ON B.ID = C.FocusID AND C.State>= 0 AND C.State<100
@@ -145,9 +133,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
         public override async Task<List<WatermeterInfo>> GetWatermeterInfoList(string gatherCode = "V4-Gather-8890")
         {
             string sql = $@"SELECT 
-                            A.ID as MeterId,
+                            A.ID,
                             A.Name,
-                            A.FocusID as FocusId,
+                            A.FocusID,
                             A.MeteringCode,
                             A.Baudrate,
                             A.MeteringPort,

src/JiShe.CollectBus.Common/Consts/RedisConst.cs

@@ -28,49 +28,25 @@ namespace JiShe.CollectBus.Common.Consts
         /// </summary>
         public const string FifteenMinuteAcquisitionTimeInterval = "Fifteen";
 
-        public const string MeterInfo = "MeterInfo";
         /// <summary>
         /// 缓存表计信息,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
         /// </summary>
-        public const string CacheMeterInfoHashKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:{"{3}"}";
+        public const string CacheMeterInfoKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:MeterInfo:{"{2}"}:{"{3}"}:";
 
-        /// <summary>
-        /// 缓存表计信息索引Set缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
-        /// </summary>
-        public const string CacheMeterInfoSetIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:SetIndex:{"{3}"}";
-
-        /// <summary>
-        /// 缓存表计信息排序索引ZSET缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
-        /// </summary>
-        public const string CacheMeterInfoZSetScoresIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{MeterInfo}:{"{2}"}:ZSetScoresIndex:{"{3}"}";
-          
-
-        public const string TaskInfo = "TaskInfo";
         /// <summary>
         /// 缓存待下发的指令生产任务数据,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
         /// </summary>
-        public const string CacheTasksToBeIssuedKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TaskInfo}:{"{2}"}:{"{3}"}";
+        public const string CacheTasksToBeIssuedKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:TaskInfo:{"{2}"}:{"{3}"}";
-
-        public const string TelemetryPacket = "TelemetryPacket";
-        /// <summary>
-        /// 缓存表计下发指令数据集,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率，{4}=>集中器所在分组，{5}=>时间格式的任务批次
-        /// </summary>
-        public const string CacheTelemetryPacketInfoHashKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:{"{3}"}:{"{4}"}:{"{5}"}";
 
         /// <summary>
-        /// 缓存表计下发指令数据集索引Set缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率，{4}=>集中器所在分组，{5}=>时间格式的任务批次
+        /// 缓存表计下发指令数据集,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率
         /// </summary>
-        public const string CacheTelemetryPacketInfoSetIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:SetIndex:{"{3}"}:{"{4}"}:{"{5}"}";
+        public const string CacheTelemetryPacketInfoKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:TelemetryPacket:{"{2}"}:{"{3}"}:";
 
         /// <summary>
-        /// 缓存表计下发指令数据集排序索引ZSET缓存Key,{0}=>系统类型，{1}=>应用服务部署标记，{2}=>表计类别，{3}=>采集频率，{4}=>集中器所在分组，{5}=>时间格式的任务批次
+        /// 缓存设备平衡关系映射结果,{0}=>系统类型，{1}=>应用服务部署标记
         /// </summary>
-        public const string CacheTelemetryPacketInfoZSetScoresIndexKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:{TelemetryPacket}:{"{2}"}:ZSetScoresIndex:{"{3}"}:{"{4}"}:{"{5}"}";
+        public const string CacheDeviceBalanceRelationMapResultKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:RelationMap";
-
-        ///// <summary>
-        ///// 缓存设备平衡关系映射结果,{0}=>系统类型，{1}=>应用服务部署标记
-        ///// </summary>
-        //public const string CacheDeviceBalanceRelationMapResultKey = $"{CacheBasicDirectoryKey}{"{0}:{1}"}:RelationMap";
 
         public const string CacheAmmeterFocusKey = "CacheAmmeterFocusKey";
     }

src/JiShe.CollectBus.Common/DeviceBalanceControl/DeviceGroupBalanceControl.cs

@@ -161,6 +161,7 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl
                 MaxDegreeOfParallelism = maxThreads.Value,
             };
 
+            TimeSpan timeSpan = TimeSpan.FromMicroseconds(5);
             await Task.Run(() =>
             {
                 Parallel.For(0, cache.CachedGroups.Length, options, async groupId =>
@@ -168,7 +169,8 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl
                     var queue = groupQueues[groupId];
                     while (queue.TryDequeue(out T item))
                     {
-                        processor(item, groupId);
+                        await Task.Delay(timeSpan);
+                        processor(item, Thread.CurrentThread.ManagedThreadId);
                     }
                 });
             });
@@ -181,14 +183,14 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl
         /// <typeparam name="T">已经分组的设备信息</typeparam>
         /// <param name="items">部分或者全部的已经分组的设备集合</param>
         /// <param name="deviceIdSelector">从泛型对象提取deviceId</param>
-        /// <param name="processor">处理委托（参数：当前对象，分组ID）</param>
+        /// <param name="processor">处理委托（参数：当前对象，线程ID）</param>
         /// <param name="maxConcurrency">可选最佳并发度</param>
         /// <returns></returns>
         /// <exception cref="InvalidOperationException"></exception>
         public static async Task ProcessWithThrottleAsync<T>(
             List<T> items,
             Func<T, string> deviceIdSelector,
-            Action<T,int> processor,
+            Action<T> processor,
             int? maxConcurrency = null)
         {
             var cache = _currentCache ?? throw new InvalidOperationException("缓存未初始化");
@@ -242,7 +244,7 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl
         /// <summary>
         /// 分组异步处理（带节流）
         /// </summary>
-        private static async Task ProcessItemAsync<T>(T item, Action<T,int> processor, int groupId)
+        private static async Task ProcessItemAsync<T>(T item, Action<T> processor, int groupId)
         {
             // 使用内存缓存降低CPU负载
             await Task.Yield(); // 立即释放当前线程
@@ -253,7 +255,7 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl
             {
                 ExecutionContext.Run(context!, state =>
                 {
-                    processor(item,groupId);
+                    processor(item);
                 }, null);
             });
         }

src/JiShe.CollectBus.Common/Extensions/EnumerableExtensions.cs

@@ -89,26 +89,5 @@ namespace JiShe.CollectBus.Common.Extensions
             if (buffer.Count > 0)
                 yield return buffer;
         }
-
-        //public static IEnumerable<IEnumerable<T>> Batch<T>(this IEnumerable<T> source, int batchSize)
-        //{
-        //    if (batchSize <= 0)
-        //        throw new ArgumentOutOfRangeException(nameof(batchSize));
-
-        //    using var enumerator = source.GetEnumerator();
-        //    while (enumerator.MoveNext())
-        //    {
-        //        yield return GetBatch(enumerator, batchSize);
-        //    }
-        //}
-
-        //private static IEnumerable<T> GetBatch<T>(IEnumerator<T> enumerator, int batchSize)
-        //{
-        //    do
-        //    {
-        //        yield return enumerator.Current;
-        //        batchSize--;
-        //    } while (batchSize > 0 && enumerator.MoveNext());
-        //}
     }
 }

src/JiShe.CollectBus.Common/JiShe.CollectBus.Common.csproj

@@ -17,7 +17,6 @@
 
 	<ItemGroup>
 		<PackageReference Include="JetBrains.Annotations" Version="2024.2.0" />
-		<PackageReference Include="Mapster" Version="7.4.0" />
 		<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="8.0.0" />
 		<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
 		<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.2" />

src/JiShe.CollectBus.Common/Models/BusCacheGlobalPagedResult.cs

@@ -1,56 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Common.Models
-{
-    /// <summary>
-    /// 缓存全局分页结果
-    /// </summary>
-    /// <typeparam name="T"></typeparam>
-    public class BusCacheGlobalPagedResult<T>
-    {
-        /// <summary>
-        /// 数据集合
-        /// </summary>
-        public List<T> Items { get; set; }
-
-        /// <summary>
-        /// 总条数
-        /// </summary>
-        public long TotalCount { get; set; }
-
-        /// <summary>
-        /// 每页条数
-        /// </summary>
-        public int PageSize { get; set; }
-
-        /// <summary>
-        /// 总页数
-        /// </summary>
-        public int PageCount
-        {
-            get
-            {
-                return (int)Math.Ceiling((double)TotalCount / PageSize);
-            }
-        }
-
-        /// <summary>
-        /// 是否有下一页
-        /// </summary>
-        public bool HasNext { get; set; }
-
-        /// <summary>
-        /// 下一页的分页索引
-        /// </summary>
-        public decimal? NextScore { get; set; }
-
-        /// <summary>
-        /// 下一页的分页索引
-        /// </summary>
-        public string NextMember { get; set; }
-    }
-}

src/JiShe.CollectBus.Common/Models/BusGlobalPagedResult.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Common.Models
+{
+    public class GlobalPagedResult<T>
+    {
+        /// <summary>
+        /// 数据集合
+        /// </summary>
+        public List<T> Items { get; set; }
+
+        /// <summary>
+        /// 是否有下一页
+        /// </summary>
+        public bool HasNext { get; set; }
+
+        /// <summary>
+        /// 下一页的分页索引
+        /// </summary>
+        public long? NextScore { get; set; }
+
+        /// <summary>
+        /// 下一页的分页索引
+        /// </summary>
+        public string NextMember { get; set; }
+    }
+}

src/JiShe.CollectBus.Common/Models/BusPagedResult.cs

@@ -31,20 +31,5 @@ namespace JiShe.CollectBus.Common.Models
         /// 数据集合
         /// </summary>
         public IEnumerable<T> Items { get; set; }
-
-        /// <summary>
-        /// 是否有下一页
-        /// </summary>
-        public bool HasNext { get; set; }
-
-        /// <summary>
-        /// 下一页的分页索引
-        /// </summary>
-        public decimal? NextScore { get; set; }
-
-        /// <summary>
-        /// 下一页的分页索引
-        /// </summary>
-        public string NextMember { get; set; }
     }
 }

src/JiShe.CollectBus.Common/Models/DeviceCacheBasicModel.cs

@@ -9,7 +9,7 @@ namespace JiShe.CollectBus.Common.Models
     /// <summary>
     /// 设备缓存基础模型
     /// </summary>
-    public abstract class DeviceCacheBasicModel
+    public class DeviceCacheBasicModel
     {
         /// <summary>
         /// 集中器Id
@@ -20,15 +20,5 @@ namespace JiShe.CollectBus.Common.Models
         /// 表Id
         /// </summary>
         public int MeterId { get; set; }
-
-        /// <summary>
-        /// 关系映射标识，用于ZSet的Member字段和Set的Value字段，具体值可以根据不同业务场景进行定义
-        /// </summary>
-        public virtual string MemberId => $"{FocusId}:{MeterId}";
-
-        /// <summary>
-        /// ZSet排序索引分数值，具体值可以根据不同业务场景进行定义，例如时间戳
-        /// </summary>
-        public virtual long ScoreValue=> ((long)FocusId << 32) | (uint)MeterId;
     }
 }

src/JiShe.CollectBus.Domain/Ammeters/AmmeterInfo.cs

@@ -1,6 +1,4 @@
-using FreeSql.DataAnnotations;
+using System;
-using JiShe.CollectBus.Common.Models;
-using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
@@ -8,25 +6,23 @@ using System.Threading.Tasks;
 
 namespace JiShe.CollectBus.Ammeters
 {
-    public class AmmeterInfo: DeviceCacheBasicModel
+    public class AmmeterInfo
     {
         /// <summary>
-        /// 关系映射标识，用于ZSet的Member字段和Set的Value字段，具体值可以根据不同业务场景进行定义
+        /// 电表ID
         /// </summary>
-        [Column(IsIgnore = true)]
+        public int ID { get; set; }
-        public override string MemberId => $"{FocusId}:{MeterId}";
-
-        /// <summary>
-        /// ZSet排序索引分数值，具体值可以根据不同业务场景进行定义，例如时间戳
-        /// </summary>
-        [Column(IsIgnore = true)]
-        public override long ScoreValue => ((long)FocusId << 32) | (uint)DateTime.Now.Ticks;
 
         /// <summary>
         /// 电表名称
         /// </summary>
         public string Name { get; set; }
 
+        /// <summary>
+        /// 集中器ID
+        /// </summary>
+        public int FocusID { get; set; }
+
         /// <summary>
         /// 集中器地址
         /// </summary>

src/JiShe.CollectBus.Domain/Ammeters/AmmeterInfoTemp.cs

@@ -1,152 +0,0 @@
-using JiShe.CollectBus.Common.Models;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Ammeters
-{
-    public class AmmeterInfoTemp 
-    {
-        /// <summary>
-        /// 集中器Id
-        /// </summary>
-        public int FocusID { get; set; }
-
-        /// <summary>
-        /// 表Id
-        /// </summary>
-        public int Id { get; set; }
-
-        /// <summary>
-        /// 电表名称
-        /// </summary>
-        public string Name { get; set; }
-
-        /// <summary>
-        /// 集中器地址
-        /// </summary>
-        public string FocusAddress { get; set; }
-
-        /// <summary>
-        /// 集中器地址
-        /// </summary>
-        public string Address { get; set; }
-
-        /// <summary>
-        /// 集中器区域代码
-        /// </summary>
-        public string AreaCode { get; set; }
-
-        /// <summary>
-        /// 电表类别 （1单相、2三相三线、3三相四线）,
-        /// 07协议： 开合闸指令(1A开闸断电,1C单相表合闸,1B多相表合闸)  645 2007 表
-        /// 97协议：//true(合闸);false(跳闸)  545 1997 没有单相多相  之分  "true" ? "9966" : "3355"
-        /// </summary>		
-        public int TypeName { get; set; }
-
-        /// <summary>
-        /// 跳合闸状态字段：  0 合闸，1 跳闸
-        /// 电表：TripState （0 合闸-通电, 1 断开、跳闸）；
-        /// </summary>
-        public int TripState { get; set; }
-
-        /// <summary>
-        /// 规约  -电表default(30) 1：97协议，30：07协议
-        /// </summary>
-        public int? Protocol { get; set; }
-
-        /// <summary>
-        /// 一个集中器下的[MeteringCode]必须唯一。 PN
-        /// </summary>
-        public int MeteringCode { get; set; }
-
-        /// <summary>
-        /// 电表通信地址
-        /// </summary>
-        public string AmmerterAddress { get; set; }
-
-        /// <summary>
-        /// 波特率 default(2400)
-        /// </summary>
-        public int Baudrate { get; set; }
-
-        /// <summary>
-        ///  MeteringPort 端口就几个可以枚举。
-        /// </summary>
-        public int MeteringPort { get; set; }
-
-        /// <summary>
-        /// 电表密码
-        /// </summary>		
-        public string Password { get; set; }
-
-        /// <summary>
-        /// 采集时间间隔(分钟，如15)
-        /// </summary>
-        public int TimeDensity { get; set; }
-
-        /// <summary>
-        /// 该电表方案下采集项，JSON格式，如：["0D_80","0D_80"]
-        /// </summary>
-        public string ItemCodes { get; set; }
-
-        /// <summary>        
-        /// State表状态: 
-        /// 0新装（未下发），1运行(档案下发成功时设置状态值1)， 2暂停, 100销表（销表后是否重新启用）
-        /// 特定：State   -1 已删除
-        /// </summary>		
-        public int State { get; set; }
-
-        /// <summary>
-        /// 是否自动采集(0:主动采集，1：自动采集)
-        /// </summary>
-        public int AutomaticReport { get; set; }
-
-        /// <summary>
-        /// 该电表方案下采集项编号
-        /// </summary>
-        public string DataTypes { get; set; }
-
-        /// <summary>
-        /// 品牌型号
-        /// </summary>
-        public string BrandType { get; set; }
-
-        /// <summary>
-        /// 采集器编号
-        /// </summary>
-        public string GatherCode { get; set; }
-
-        /// <summary>
-        /// 是否特殊表，1是特殊电表
-        /// </summary>
-        public int Special { get; set; }
-
-        /// <summary>
-        /// 费率类型，单、多   (SingleRate :单费率（单相表1），多费率（其他0） ，与TypeName字段无关)  
-        /// SingleRate ? "单" : "复"  
-        /// [SingleRate]  --0 复费率 false ， 1 单费率 true   （与PayPlanID保持一致）
-        ///对应 TB_PayPlan.Type: 1复费率，2单费率
-        /// </summary>
-        public bool SingleRate { get; set; }
-
-        /// <summary>
-        /// 项目ID
-        /// </summary>
-        public int ProjectID { get; set; }
-
-        /// <summary>
-        /// 数据库业务ID
-        /// </summary>
-        public int DatabaseBusiID { get; set; }
-
-        /// <summary>
-        /// 是否异常集中器 0:正常，1异常
-        /// </summary>
-        public int AbnormalState { get; set; }
-
-        public DateTime LastTime { get; set; } 
-    }
-}

src/JiShe.CollectBus.Domain/IotSystems/MeterReadingRecords/MeterReadingTelemetryPacketInfo.cs

@@ -1,141 +0,0 @@
-using JiShe.CollectBus.Common.Enums;
-using JiShe.CollectBus.Common.Models;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using Volo.Abp.Domain.Entities;
-using Volo.Abp.Domain.Entities.Auditing;
-
-namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
-{
-    /// <summary>
-    /// 抄读任务Redis缓存数据记录
-    /// </summary>
-    public class MeterReadingTelemetryPacketInfo : DeviceCacheBasicModel
-    {
-        /// <summary>
-        /// 关系映射标识，用于ZSet的Member字段和Set的Value字段，具体值可以根据不同业务场景进行定义
-        /// </summary>
-        public override string MemberId => $"{FocusId}:{MeterId}:{ItemCode}";
-
-        /// <summary>
-        /// ZSet排序索引分数值，具体值可以根据不同业务场景进行定义，例如时间戳
-        /// </summary>
-        public override long ScoreValue => ((long)FocusId << 32) | (uint)DateTime.Now.Ticks;
-
-
-        /// <summary>
-        /// 是否手动操作
-        /// </summary>
-        public bool ManualOrNot { get; set; }
-
-        /// <summary>
-        /// 任务数据唯一标记
-        /// </summary>
-        public string TaskMark { get; set; }
-
-        /// <summary>
-        /// 时间戳标记，IoTDB时间列处理，上报通过构建标记获取唯一标记匹配时间戳。
-        /// </summary>
-        public long Timestamps { get; set; }
-
-        /// <summary>
-        /// 是否超时
-        /// </summary>
-        public bool IsTimeout { get; set; } = false;
-
-        /// <summary>
-        /// 待抄读时间
-        /// </summary>
-        public DateTime PendingCopyReadTime { get; set; }
-         
-        
-        /// <summary>
-        /// 集中器地址
-        /// </summary>
-        public string FocusAddress { get; set; }
- 
-        /// <summary>
-        /// 表地址
-        /// </summary>
-        public string MeterAddress { get; set; }
-
-        /// <summary>
-        /// 表类型
-        /// </summary>
-        public MeterTypeEnum MeterType { get; set; }
-
-        /// <summary>
-        /// 项目ID
-        /// </summary>
-        public int ProjectID { get; set; }
-
-        /// <summary>
-        /// 数据库业务ID
-        /// </summary>
-        public int DatabaseBusiID { get; set; }
-
-        /// <summary>
-        /// AFN功能码
-        /// </summary>
-        public AFN AFN { get; set; }
-
-        /// <summary>
-        /// 抄读功能码
-        /// </summary>
-        public int Fn { get; set; }
-
-        /// <summary>
-        /// 抄读计量点
-        /// </summary>
-        public int Pn { get; set; }
-
-        /// <summary>
-        /// 采集项编码
-        /// </summary>
-        public string ItemCode { get; set;}
-         
-
-        /// <summary>
-        /// 创建时间
-        /// </summary>
-        public DateTime CreationTime { get; set; }
-
-        /// <summary>
-        /// 下发消息内容
-        /// </summary>
-        public string IssuedMessageHexString { get; set; }
-
-        /// <summary>
-        /// 下发消息Id
-        /// </summary>
-        public string IssuedMessageId { get; set; }
-
-        /// <summary>
-        /// 消息上报内容
-        /// </summary>
-        public string? ReceivedMessageHexString { get; set; }
-
-        /// <summary>
-        /// 消息上报时间
-        /// </summary>
-        public DateTime? ReceivedTime { get; set; }
-
-        /// <summary>
-        /// 上报消息Id
-        /// </summary>
-        public string ReceivedMessageId { get; set; } 
-
-        /// <summary>
-        /// 上报报文解析备注，异常情况下才有
-        /// </summary>
-        public string ReceivedRemark { get; set; }
-         
-        //public void CreateDataId(Guid Id)
-        //{
-        //    this.Id = Id;
-        //}
-    }
-}

src/JiShe.CollectBus.Domain/IotSystems/Watermeter/WatermeterInfo.cs

@@ -1,5 +1,4 @@
 using JiShe.CollectBus.Common.Enums;
-using JiShe.CollectBus.Common.Models;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -11,8 +10,13 @@ namespace JiShe.CollectBus.IotSystems.Watermeter
     /// <summary>
     /// 水表信息
     /// </summary>
-    public class WatermeterInfo: DeviceCacheBasicModel
+    public class WatermeterInfo
     {
+        /// <summary>
+        /// 水表ID
+        /// </summary>
+        public int ID { get; set; }
+
         /// <summary>
         /// 水表名称
         /// </summary>
@@ -22,6 +26,11 @@ namespace JiShe.CollectBus.IotSystems.Watermeter
         /// </summary>		
         public string Password { get; set; }
 
+        /// <summary>
+        /// 集中器ID
+        /// </summary>
+        public int FocusID { get; set; }
+
         /// <summary>
         /// 集中器地址
         /// </summary>

src/JiShe.CollectBus.FreeRedisProvider/FreeRedisProvider.cs

@@ -1,14 +1,11 @@
 using FreeRedis;
 using JiShe.CollectBus.Common.Helpers;
 using JiShe.CollectBus.Common.Models;
-using JiShe.CollectBus.Common.Extensions;
 using JiShe.CollectBus.FreeRedisProvider.Options; 
 using Microsoft.Extensions.Options;
 using System.Diagnostics;
 using System.Text.Json;
 using Volo.Abp.DependencyInjection;
-using static System.Runtime.InteropServices.JavaScript.JSType;
-using System.Collections.Concurrent;
 
 namespace JiShe.CollectBus.FreeRedisProvider
 {
@@ -36,7 +33,7 @@ namespace JiShe.CollectBus.FreeRedisProvider
         public IRedisClient GetInstance()
         {
 
-            var connectionString = $"{_option.Configuration},defaultdatabase={_option.DefaultDB},MaxPoolSize={_option.MaxPoolSize}";
+            var connectionString = $"{_option.Configuration},defaultdatabase={_option.DefaultDB}";
             Instance = new RedisClient(connectionString);
             Instance.Serialize = obj => BusJsonSerializer.Serialize(obj);
             Instance.Deserialize = (json, type) => BusJsonSerializer.Deserialize(json, type);
@@ -44,464 +41,442 @@ namespace JiShe.CollectBus.FreeRedisProvider
             return Instance;
         }
 
-        ///// <summary>
+
-        ///// 单个添加数据
+        //public async Task AddMeterZSetCacheData<T>(string redisCacheKey, string redisCacheIndexKey, decimal score, T data)
-        ///// </summary>
-        ///// <typeparam name="T"></typeparam>
-        ///// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
-        ///// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
-        ///// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
-        ///// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
-        ///// <param name="data">表计信息</param>
-        ///// <param name="timestamp">可选时间戳</param>
-        ///// <returns></returns>
-        //public async Task AddMeterCacheData<T>(
-        //string redisCacheKey,
-        //string redisCacheFocusIndexKey,
-        //string redisCacheScoresIndexKey,
-        //string redisCacheGlobalIndexKey,
-        //T data,
-        //DateTimeOffset? timestamp = null) where T : DeviceCacheBasicModel
         //{
-        //    // 参数校验增强
+        //    if (score < 0 || data == null || string.IsNullOrWhiteSpace(redisCacheKey) || string.IsNullOrWhiteSpace(redisCacheIndexKey))
-        //    if (data == null || string.IsNullOrWhiteSpace(redisCacheKey)
-        //        || string.IsNullOrWhiteSpace(redisCacheFocusIndexKey)
-        //        || string.IsNullOrWhiteSpace(redisCacheScoresIndexKey)
-        //        || string.IsNullOrWhiteSpace(redisCacheGlobalIndexKey))
         //    {
-        //        throw new ArgumentException($"{nameof(AddMeterCacheData)} 参数异常，-101");
+        //        throw new Exception($"{nameof(AddMeterZSetCacheData)} 参数异常，-101");
         //    }
 
-        //    // 计算组合score（分类ID + 时间戳）
+        //    // 生成唯一member标识
-        //    var actualTimestamp = timestamp ?? DateTimeOffset.UtcNow;
+        //    var member = data.Serialize();
 
-        //    long scoreValue = ((long)data.FocusId << 32) | (uint)actualTimestamp.Ticks;
+        //    // 计算score范围
+        //    decimal dataScore = (long)score << 32;
 
-        //    //全局索引写入
+        //    //// 事务操作
-        //    long globalScore = actualTimestamp.ToUnixTimeMilliseconds();
+        //    //using (var tran = FreeRedisProvider.Instance.Multi())
+        //    //{
+        //    //    await tran.ZAddAsync(cacheKey, score,member);
+        //    //    await tran.SAddAsync($"cat_index:{categoryId}", member);
+        //    //    object[] ret = tran.Exec();
+        //    //}
 
-        //    // 使用事务保证原子性
-        //    using (var trans = Instance.Multi())
-        //    {
-        //        // 主数据存储Hash
-        //        trans.HSet(redisCacheKey, data.MemberID, data.Serialize());
-
-        //        // 分类索引
-        //        trans.SAdd(redisCacheFocusIndexKey, data.MemberID);
-
-        //        // 排序索引使用ZSET
-        //        trans.ZAdd(redisCacheScoresIndexKey, scoreValue, data.MemberID);
-
-        //        //全局索引
-        //        trans.ZAdd(redisCacheGlobalIndexKey, globalScore, data.MemberID);
-
-        //        var results = trans.Exec();
-
-        //        if (results == null || results.Length <= 0)
-        //            throw new Exception($"{nameof(AddMeterCacheData)} 事务提交失败，-102");
-        //    }
-
-        //    await Task.CompletedTask;
-        //}
-
-        ///// <summary>
-        ///// 批量添加数据
-        ///// </summary>
-        ///// <typeparam name="T"></typeparam>
-        ///// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
-        ///// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
-        ///// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
-        ///// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
-        ///// <param name="items">数据集合</param>
-        ///// <param name="timestamp">可选时间戳</param>
-        ///// <returns></returns>
-        //public async Task BatchAddMeterData<T>(
-        //string redisCacheKey,
-        //string redisCacheFocusIndexKey,
-        //string redisCacheScoresIndexKey,
-        //string redisCacheGlobalIndexKey,
-        //IEnumerable<T> items,
-        //DateTimeOffset? timestamp = null) where T : DeviceCacheBasicModel
-        //{
-        //    if (items == null 
-        //       || items.Count() <=0
-        //       || string.IsNullOrWhiteSpace(redisCacheKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheFocusIndexKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheScoresIndexKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheGlobalIndexKey))
-        //    {
-        //        throw new ArgumentException($"{nameof(BatchAddMeterData)} 参数异常，-101");
-        //    }
-
-        //    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)
+        //        pipe.ZAdd(redisCacheKey, dataScore, member);
-        //                {
+        //        pipe.SAdd(redisCacheIndexKey, member);
-        //                    // 计算组合score（分类ID + 时间戳）
+        //        object[] ret = pipe.EndPipe();
-        //                    var actualTimestamp = timestamp ?? DateTimeOffset.UtcNow;
-
-        //                    long scoreValue = ((long)item.FocusId << 32) | (uint)actualTimestamp.Ticks;
-
-        //                    //全局索引写入
-        //                    long globalScore = actualTimestamp.ToUnixTimeMilliseconds();
-
-        //                    // 主数据存储Hash
-        //                    pipe.HSet(redisCacheKey, item.MemberID, item.Serialize());
-
-        //                    // 分类索引Set
-        //                    pipe.SAdd(redisCacheFocusIndexKey, item.MemberID);
-
-        //                    // 排序索引使用ZSET
-        //                    pipe.ZAdd(redisCacheScoresIndexKey, scoreValue, item.MemberID);
-
-        //                    //全局索引
-        //                    pipe.ZAdd(redisCacheGlobalIndexKey, globalScore, item.MemberID);
-        //                }
-        //                pipe.EndPipe();
-        //            }
-        //            semaphore.Release();
-        //        });
         //    }
 
         //    await Task.CompletedTask;
         //}
 
-        ///// <summary>
+        //public async Task<BusPagedResult<T>> GetMeterZSetPagedData<T>(
-        ///// 删除指定redis缓存key的缓存数据
-        ///// </summary>
-        ///// <typeparam name="T"></typeparam>
-        ///// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
-        ///// <param name="redisCacheFocusIndexKey">集中器索引Set缓存Key</param>
-        ///// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
-        ///// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
-        ///// <param name="data">表计信息</param>
-        ///// <returns></returns>
-        //public async Task RemoveMeterData<T>(
         //string redisCacheKey,
-        //string redisCacheFocusIndexKey,
+        //string redisCacheIndexKey,
-        //string redisCacheScoresIndexKey,
+        //decimal score,
-        //string redisCacheGlobalIndexKey,
-        //T data) where T : DeviceCacheBasicModel
-        //{
-
-        //    if (data == null 
-        //       || string.IsNullOrWhiteSpace(redisCacheKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheFocusIndexKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheScoresIndexKey)
-        //       || string.IsNullOrWhiteSpace(redisCacheGlobalIndexKey))
-        //    {
-        //        throw new ArgumentException($"{nameof(RemoveMeterData)} 参数异常，-101");
-        //    }
-
-        //    const string luaScript = @"
-        //        local mainKey = KEYS[1]
-        //        local focusIndexKey = KEYS[2]
-        //        local scoresIndexKey = KEYS[3]
-        //        local globalIndexKey = KEYS[4]
-        //        local member = ARGV[1]
-
-        //        local deleted = 0
-        //        if redis.call('HDEL', mainKey, member) > 0 then
-        //            deleted = 1
-        //        end
-
-        //        redis.call('SREM', focusIndexKey, member)
-        //        redis.call('ZREM', scoresIndexKey, member)
-        //        redis.call('ZREM', globalIndexKey, member)
-        //        return deleted
-        //    ";
-
-        //    var keys = new[]
-        //    {
-        //        redisCacheKey,
-        //        redisCacheFocusIndexKey,
-        //        redisCacheScoresIndexKey,
-        //        redisCacheGlobalIndexKey
-        //    };
-
-        //    var result = await Instance.EvalAsync(luaScript, keys, new[] { data.MemberID });
-
-        //    if ((int)result == 0)
-        //        throw new KeyNotFoundException("指定数据不存在");
-        //}
-
-        ///// <summary>
-        ///// 修改表计缓存信息
-        ///// </summary>
-        ///// <typeparam name="T"></typeparam>
-        ///// <param name="redisCacheKey">主数据存储Hash缓存Key</param>
-        ///// <param name="oldRedisCacheFocusIndexKey">旧集中器索引Set缓存Key</param>
-        ///// <param name="newRedisCacheFocusIndexKey">新集中器索引Set缓存Key</param>
-        ///// <param name="redisCacheScoresIndexKey">集中器排序索引ZSET缓存Key</param>
-        ///// <param name="redisCacheGlobalIndexKey">集中器采集频率分组全局索引ZSet缓存Key</param>
-        ///// <param name="newData">表计信息</param>
-        ///// <param name="newTimestamp">可选时间戳</param>
-        ///// <returns></returns>
-        //public async Task UpdateMeterData<T>(
-        //string redisCacheKey, 
-        //string oldRedisCacheFocusIndexKey,
-        //string newRedisCacheFocusIndexKey,
-        //string redisCacheScoresIndexKey,
-        //string redisCacheGlobalIndexKey,
-        //T newData,
-        //DateTimeOffset? newTimestamp = null) where T : DeviceCacheBasicModel
-        //{
-        //    if (newData == null
-        //      || string.IsNullOrWhiteSpace(redisCacheKey)
-        //      || string.IsNullOrWhiteSpace(oldRedisCacheFocusIndexKey)
-        //      || string.IsNullOrWhiteSpace(newRedisCacheFocusIndexKey)
-        //      || string.IsNullOrWhiteSpace(redisCacheScoresIndexKey)
-        //      || string.IsNullOrWhiteSpace(redisCacheGlobalIndexKey))
-        //    {
-        //        throw new ArgumentException($"{nameof(UpdateMeterData)} 参数异常，-101");
-        //    }
-
-        //    var luaScript = @"
-        //        local mainKey = KEYS[1]
-        //        local oldFocusIndexKey = KEYS[2]
-        //        local newFocusIndexKey = KEYS[3]
-        //        local scoresIndexKey = KEYS[4]
-        //        local globalIndexKey = KEYS[5]
-        //        local member = ARGV[1]
-        //        local newData = ARGV[2]
-        //        local newScore = ARGV[3]
-        //        local newGlobalScore = ARGV[4]
-
-        //        -- 校验存在性
-        //        if redis.call('HEXISTS', mainKey, member) == 0 then
-        //            return 0
-        //        end
-
-        //        -- 更新主数据
-        //        redis.call('HSET', mainKey, member, newData)
-
-        //        -- 处理变更
-        //        if newScore ~= '' then
-        //            -- 删除旧索引
-        //            redis.call('SREM', oldFocusIndexKey, member)
-        //            redis.call('ZREM', scoresIndexKey, member)
-            
-        //            -- 添加新索引
-        //            redis.call('SADD', newFocusIndexKey, member)
-        //            redis.call('ZADD', scoresIndexKey, newScore, member)
-        //        end
-
-        //        -- 更新全局索引
-        //        if newGlobalScore ~= '' then
-        //            -- 删除旧索引
-        //            redis.call('ZREM', globalIndexKey, member)
-
-        //            -- 添加新索引
-        //            redis.call('ZADD', globalIndexKey, newGlobalScore, member)
-        //        end
-
-        //        return 1
-        //    ";
-
-        //    var actualTimestamp = newTimestamp ?? DateTimeOffset.UtcNow;
-        //    var newGlobalScore = actualTimestamp.ToUnixTimeMilliseconds();
-        //    var newScoreValue = ((long)newData.FocusId << 32) | (uint)actualTimestamp.Ticks;
-
-        //    var result = await Instance.EvalAsync(luaScript,
-        //        new[]
-        //        {
-        //            redisCacheKey,
-        //            oldRedisCacheFocusIndexKey,
-        //            newRedisCacheFocusIndexKey,
-        //            redisCacheScoresIndexKey,
-        //            redisCacheGlobalIndexKey
-        //        },
-        //        new object[]
-        //        {
-        //            newData.MemberID,
-        //            newData.Serialize(),
-        //            newScoreValue.ToString() ?? "",
-        //            newGlobalScore.ToString() ?? ""
-        //        });
-
-        //    if ((int)result == 0)
-        //    {
-        //        throw new KeyNotFoundException($"{nameof(UpdateMeterData)}指定Key{redisCacheKey}的数据不存在");
-        //    }
-        //}
-         
-        //public async Task<BusPagedResult<T>> SingleGetMeterPagedData<T>(
-        //string redisCacheKey,
-        //string redisCacheScoresIndexKey,
-        //int focusId,
         //int pageSize = 10,
-        //int pageIndex = 1,
+        //int pageIndex = 1)
-        //bool descending = true)
         //{
-        //    // 计算score范围
+        //    if (score < 0 || string.IsNullOrWhiteSpace(redisCacheKey) || string.IsNullOrWhiteSpace(redisCacheIndexKey))
-        //    long minScore = (long)focusId << 32;
+        //    {
-        //    long maxScore = ((long)focusId + 1) << 32;
+        //        throw new Exception($"{nameof(GetMeterZSetPagedData)} 参数异常，-101");
+        //    }
 
-        //    // 分页参数计算
+        //    // 计算score范围
+        //    decimal minScore = (long)score << 32;
+        //    decimal maxScore = ((long)score + 1) << 32;
+
+        //    // 分页参数
         //    int start = (pageIndex - 1) * pageSize;
 
-        //    // 获取排序后的member列表
+        //    // 查询主数据
-        //    var members = descending
+        //    var members = await Instance.ZRevRangeByScoreAsync(
-        //        ? await Instance.ZRevRangeByScoreAsync(
+        //        redisCacheKey,
-        //            redisCacheScoresIndexKey,
         //        maxScore,
         //        minScore,
-        //            start,
+        //        offset: start,
-        //            pageSize)
+        //        count: pageSize
-        //        : await Instance.ZRangeByScoreAsync(
+        //    );
-        //            redisCacheScoresIndexKey,
-        //            minScore,
-        //            maxScore,
-        //            start,
-        //            pageSize);
 
-        //    // 批量获取实际数据
+        //    if (members == null)
-        //    var dataTasks = members.Select(m =>
+        //    {
-        //        Instance.HGetAsync<T>(redisCacheKey, m)).ToArray();
+        //        throw new Exception($"{nameof(GetMeterZSetPagedData)} 获取缓存的信息失败，第 {pageIndex + 1} 页数据未返回，-102");
-        //    await Task.WhenAll(dataTasks);
+        //    }
 
-        //    // 总数统计优化
+        //    // 查询总数
-        //    var total = await Instance.ZCountAsync(
+        //    var total = await Instance.ZCountAsync(redisCacheKey, minScore, maxScore);
-        //        redisCacheScoresIndexKey,
-        //        minScore,
-        //        maxScore);
 
         //    return new BusPagedResult<T>
         //    {
-        //        Items = dataTasks.Select(t => t.Result).ToList(),
+        //        Items = members.Select(m =>
+        //            BusJsonSerializer.Deserialize<T>(m)!).ToList(),
         //        TotalCount = total,
         //        PageIndex = pageIndex,
         //        PageSize = pageSize
         //    };
         //}
 
-
+        ///// <summary>
-        //public async Task<BusPagedResult<T>> GetFocusPagedData<T>(
+        ///// 删除数据示例
+        ///// </summary>
+        ///// <typeparam name="T"></typeparam>
+        ///// <param name="redisCacheKey">分类</param>
+        ///// <param name="redisCacheIndexKey"></param>
+        ///// <param name="data"></param>
+        ///// <returns></returns>
+        //public async Task RemoveMeterZSetData<T>(
         //string redisCacheKey,
-        //string redisCacheScoresIndexKey,
+        //string redisCacheIndexKey,
-        //int focusId,
+        //T data)
-        //int pageSize = 10,
-        //long? lastScore = null,
-        //string lastMember = null,
-        //bool descending = true) where T : DeviceCacheBasicModel
         //{
-        //    // 计算分数范围
-        //    long minScore = (long)focusId << 32;
-        //    long maxScore = ((long)focusId + 1) << 32;
 
-        //    // 获取成员列表
+        //    // 查询需要删除的member
-        //    var members = await GetSortedMembers(
+        //    var members = await Instance.SMembersAsync(redisCacheIndexKey);
-        //        redisCacheScoresIndexKey,
+        //    var target = members.FirstOrDefault(m =>
-        //        minScore,
+        //        BusJsonSerializer.Deserialize<T>(m) == data);//泛型此处该如何处理？
-        //        maxScore,
-        //        pageSize,
-        //        lastScore,
-        //        lastMember,
-        //        descending);
 
-        //    // 批量获取数据
+        //    if (target != null)
-        //    var dataDict = await Instance.HMGetAsync<T>(redisCacheKey, members.CurrentItems);
-
-        //    return new BusPagedResult<T>
         //    {
-        //        Items = dataDict,
+        //        using (var trans = Instance.Multi())
-        //        TotalCount = await GetTotalCount(redisCacheScoresIndexKey, minScore, maxScore),
+        //        {
-        //        HasNext = members.HasNext,
+        //            trans.ZRem(redisCacheKey, target);
-        //        NextScore = members.NextScore,
+        //            trans.SRem(redisCacheIndexKey, target);
-        //        NextMember = members.NextMember
+        //            trans.Exec();
-        //    };
+        //        }
-        //}
+        //    }
 
-        //private async Task<(string[] CurrentItems, bool HasNext, decimal? NextScore, string NextMember)>
+        //    await Task.CompletedTask;
-        //    GetSortedMembers(
-        //        string zsetKey,
-        //        long minScore,
-        //        long maxScore,
-        //        int pageSize,
-        //        long? lastScore,
-        //        string lastMember,
-        //        bool descending)
-        //{
-        //    var querySize = pageSize + 1;
-        //    var (startScore, exclude) = descending
-        //        ? (lastScore ?? maxScore, lastMember)
-        //        : (lastScore ?? minScore, lastMember);
-
-        //    var members = descending
-        //        ? await Instance.ZRevRangeByScoreAsync(
-        //            zsetKey,
-        //            max: startScore,
-        //            min: minScore,
-        //            offset: 0,
-        //            count: querySize)
-        //        : await Instance.ZRangeByScoreAsync(
-        //            zsetKey,
-        //            min: startScore,
-        //            max: maxScore,
-        //            offset: 0,
-        //            count: querySize);
-
-        //    var hasNext = members.Length > pageSize;
-        //    var currentItems = members.Take(pageSize).ToArray();
-
-        //    var nextCursor = currentItems.Any()
-        //        ? await GetNextCursor(zsetKey, currentItems.Last(), descending)
-        //        : (null, null);
-
-        //    return (currentItems, hasNext, nextCursor.score, nextCursor.member);
-        //}
-
-        //private async Task<long> GetTotalCount(string zsetKey, long min, long max)
-        //{
-        //    // 缓存计数优化
-        //    var cacheKey = $"{zsetKey}_count_{min}_{max}";
-        //    var cached = await Instance.GetAsync<long?>(cacheKey);
-
-        //    if (cached.HasValue)
-        //        return cached.Value;
-
-        //    var count = await Instance.ZCountAsync(zsetKey, min, max);
-        //    await Instance.SetExAsync(cacheKey, 60, count); // 缓存60秒
-        //    return count;
         //}
 
 
-        //public async Task<Dictionary<int, BusPagedResult<T>>> BatchGetMeterPagedData<T>(
+        public async Task AddMeterZSetCacheData<T>(
-        //string redisCacheKey,
+        string redisCacheKey,
-        //string redisCacheScoresIndexKey,
+        string redisCacheIndexKey,
-        //IEnumerable<int> focusIds,
+        int categoryId, // 新增分类ID参数
-        //int pageSizePerFocus = 10) where T : DeviceCacheBasicModel
+        T data,
+        DateTimeOffset? timestamp = null)
+        {
+            // 参数校验增强
+            if (data == null || string.IsNullOrWhiteSpace(redisCacheKey)
+                || string.IsNullOrWhiteSpace(redisCacheIndexKey))
+            {
+                throw new ArgumentException("Invalid parameters");
+            }
+
+            // 生成唯一member标识（带数据指纹）
+            var member = $"{categoryId}:{Guid.NewGuid()}";
+            var serializedData = data.Serialize();
+
+            // 计算组合score（分类ID + 时间戳）
+            var actualTimestamp = timestamp ?? DateTimeOffset.UtcNow;
+
+            long scoreValue = ((long)categoryId << 32) | (uint)actualTimestamp.Ticks;
+            
+            //全局索引写入
+            long globalScore = actualTimestamp.ToUnixTimeMilliseconds();
+
+            // 使用事务保证原子性
+            using (var trans = Instance.Multi())
+            {
+                // 主数据存储Hash
+                trans.HSet(redisCacheKey, member, serializedData);
+
+                // 排序索引使用ZSET
+                trans.ZAdd($"{redisCacheKey}_scores", scoreValue, member);
+
+                // 分类索引
+                trans.SAdd(redisCacheIndexKey, member);
+
+                //全局索引
+                trans.ZAdd("global_data_all", globalScore, member);
+
+                var results = trans.Exec();
+
+                if (results == null || results.Length <= 0)
+                    throw new Exception("Transaction failed");
+            }
+
+            await Task.CompletedTask;
+        }
+
+        public async Task BatchAddMeterData<T>(
+        string redisCacheKey,
+        string indexKey,
+        IEnumerable<T> items) where T : DeviceCacheBasicModel
+        {
+            const int BATCH_SIZE = 1000; // 每批1000条
+            var semaphore = new SemaphoreSlim(Environment.ProcessorCount * 2);
+
+            //foreach (var batch in items.Batch(BATCH_SIZE))
             //{
-        //    var results = new ConcurrentDictionary<int, BusPagedResult<T>>();
+            //    await semaphore.WaitAsync();
-        //    var parallelOptions = new ParallelOptions
-        //    {
-        //        MaxDegreeOfParallelism = Environment.ProcessorCount * 2
-        //    };
 
-        //    await Parallel.ForEachAsync(focusIds, parallelOptions, async (focusId, _) =>
+            //    _ = Task.Run(async () =>
             //    {
-        //        var data = await SingleGetMeterPagedData<T>(
+            //        using (var pipe = FreeRedisProvider.Instance.StartPipe())
-        //            redisCacheKey,
+            //        {
-        //            redisCacheScoresIndexKey,
+            //            foreach (var item in batch)
-        //            focusId,
+            //            {
-        //            pageSizePerFocus);
+            //                var member = $"{item.CategoryId}:{Guid.NewGuid()}";
+            //                long score = ((long)item.CategoryId << 32) | (uint)item.Timestamp.Ticks;
 
-        //        results.TryAdd(focusId, data);
+            //                // Hash主数据
+            //                pipe.HSet(redisCacheKey, member, item.Data.Serialize());
+
+            //                // 分类索引
+            //                pipe.ZAdd($"{redisCacheKey}_scores", score, member);
+
+            //                // 全局索引
+            //                pipe.ZAdd("global_data_all", item.Timestamp.ToUnixTimeMilliseconds(), member);
+
+            //                // 分类快速索引
+            //                pipe.SAdd(indexKey, member);
+            //            }
+            //            pipe.EndPipe();
+            //        }
+            //        semaphore.Release();
             //    });
-
-        //    return new Dictionary<int, BusPagedResult<T>>(results);
             //}
 
+            await Task.CompletedTask;
+        }
+
+        public async Task UpdateMeterData<T>(
+        string redisCacheKey,
+        string oldCategoryIndexKey,
+        string newCategoryIndexKey,
+        string memberId, // 唯一标识（格式："分类ID:GUID"）
+        T newData,
+        int? newCategoryId = null,
+        DateTimeOffset? newTimestamp = null)
+        {
+            // 参数校验
+            if (string.IsNullOrWhiteSpace(memberId))
+                throw new ArgumentException("Invalid member ID");
+
+            var luaScript = @"
+                local mainKey = KEYS[1]
+                local scoreKey = KEYS[2]
+                local oldIndex = KEYS[3]
+                local newIndex = KEYS[4]
+                local member = ARGV[1]
+                local newData = ARGV[2]
+                local newScore = ARGV[3]
+
+                -- 校验旧数据是否存在
+                if redis.call('HEXISTS', mainKey, member) == 0 then
+                    return 0
+                end
+
+                -- 更新主数据
+                redis.call('HSET', mainKey, member, newData)
+
+                -- 处理分类变更
+                if newScore ~= '' then
+                    -- 删除旧索引
+                    redis.call('SREM', oldIndex, member)
+                    -- 更新排序分数
+                    redis.call('ZADD', scoreKey, newScore, member)
+                    -- 添加新索引
+                    redis.call('SADD', newIndex, member)
+                end
+
+                return 1
+            ";
+
+            // 计算新score（当分类或时间变化时）
+            long? newScoreValue = null;
+            if (newCategoryId.HasValue || newTimestamp.HasValue)
+            {
+                var parts = memberId.Split(':');
+                var oldCategoryId = int.Parse(parts[0]);
+
+                var actualCategoryId = newCategoryId ?? oldCategoryId;
+                var actualTimestamp = newTimestamp ?? DateTimeOffset.UtcNow;
+
+                newScoreValue = ((long)actualCategoryId << 32) | (uint)actualTimestamp.Ticks;
+            }
+
+            var result = await Instance.EvalAsync(luaScript,
+                new[]
+                {
+                    redisCacheKey,
+                    $"{redisCacheKey}_scores",
+                    oldCategoryIndexKey,
+                    newCategoryIndexKey
+                },
+                new[]
+                {
+                    memberId,
+                    newData.Serialize(),
+                    newScoreValue?.ToString() ?? ""
+                });
+
+            // 如果时间戳变化则更新全局索引
+            if (newTimestamp.HasValue)
+            {
+                long newGlobalScore = newTimestamp.Value.ToUnixTimeMilliseconds();
+                await Instance.ZAddAsync("global_data_all", newGlobalScore, memberId);
+            }
+
+            if ((int)result == 0)
+                throw new KeyNotFoundException("指定数据不存在");
+        }
 
 
+        public async Task<BusPagedResult<T>> GetMeterZSetPagedData<T>(
+        string redisCacheKey,
+        string redisCacheIndexKey,
+        int categoryId,
+        int pageSize = 10,
+        int pageIndex = 1,
+        bool descending = true)
+        {
+            // 计算score范围
+            long minScore = (long)categoryId << 32;
+            long maxScore = ((long)categoryId + 1) << 32;
+
+            // 分页参数计算
+            int start = (pageIndex - 1) * pageSize;
+
+            // 获取排序后的member列表
+            var members = descending
+                ? await Instance.ZRevRangeByScoreAsync(
+                    $"{redisCacheKey}_scores",
+                    maxScore,
+                    minScore,
+                    start,
+                    pageSize)
+                : await Instance.ZRangeByScoreAsync(
+                    $"{redisCacheKey}_scores",
+                    minScore,
+                    maxScore,
+                    start,
+                    pageSize);
+
+            // 批量获取实际数据
+            var dataTasks = members.Select(m =>
+                Instance.HGetAsync<T>(redisCacheKey, m)).ToArray();
+            await Task.WhenAll(dataTasks);
+
+            // 总数统计优化
+            var total = await Instance.ZCountAsync(
+                $"{redisCacheKey}_scores",
+                minScore,
+                maxScore);
+
+            return new BusPagedResult<T>
+            {
+                Items = dataTasks.Select(t => t.Result).ToList(),
+                TotalCount = total,
+                PageIndex = pageIndex,
+                PageSize = pageSize
+            };
+        }
+
+
+        public async Task RemoveMeterZSetData<T>(
+        string redisCacheKey,
+        string redisCacheIndexKey,
+        string uniqueId) // 改为基于唯一标识删除
+        {
+            // 原子操作
+            var luaScript = @"
+                local mainKey = KEYS[1]
+                local scoreKey = KEYS[2]
+                local indexKey = KEYS[3]
+                local member = ARGV[1]
+        
+                redis.call('HDEL', mainKey, member)
+                redis.call('ZREM', scoreKey, member)
+                redis.call('SREM', indexKey, member)
+                return 1
+            ";
+
+            var keys = new[]
+            {
+                redisCacheKey,
+                $"{redisCacheKey}_scores",
+                redisCacheIndexKey
+            };
+
+            var result = await Instance.EvalAsync(luaScript,
+                keys,
+                new[] { uniqueId });
+
+            if ((int)result != 1)
+                throw new Exception("删除操作失败");
+        }
+
+        public async Task<GlobalPagedResult<T>> GetGlobalPagedData<T>(
+        string redisCacheKey,
+        int pageSize = 10,
+        long? lastScore = null,
+        string lastMember = null,
+        bool descending = true)
+        {
+            const string zsetKey = "global_data_all";
+
+            // 分页参数处理
+            var (startScore, excludeMember) = descending
+                ? (lastScore ?? long.MaxValue, lastMember)
+                : (lastScore ?? 0, lastMember);
+
+            // 获取成员列表
+            string[] members;
+            if (descending)
+            {
+                members = await Instance.ZRevRangeByScoreAsync(
+                    zsetKey,
+                    max: startScore,
+                    min: 0,
+                    offset: 0,
+                    count: pageSize + 1);
+            }
+            else
+            {
+                members = await Instance.ZRangeByScoreAsync(
+                    zsetKey,
+                    min: startScore,
+                    max: long.MaxValue,
+                    offset: 0,
+                    count: pageSize + 1);
+            }
+
+            // 处理分页结果
+            bool hasNext = members.Length > pageSize;
+            var actualMembers = members.Take(pageSize).ToArray();
+
+            // 批量获取数据（优化版本）
+            var dataTasks = actualMembers
+                .Select(m => Instance.HGetAsync<T>(redisCacheKey, m))
+                .ToArray();
+            await Task.WhenAll(dataTasks);
+
+            // 获取下一页游标
+            (long? nextScore, string nextMember) = actualMembers.Any()
+                ? await GetNextCursor(zsetKey, actualMembers.Last(), descending)
+                : (null, null);
+
+            return new GlobalPagedResult<T>
+            {
+                Items = dataTasks.Select(t => t.Result).ToList(),
+                HasNext = hasNext,
+                NextScore = nextScore,
+                NextMember = nextMember
+            };
+        }
+
+        private async Task<(long? score, string member)> GetNextCursor(
+            string zsetKey,
+            string lastMember,
+            bool descending)
+        {
+            var score = await Instance.ZScoreAsync(zsetKey, lastMember);
+            return (score.HasValue ? (long)score.Value : null, lastMember);
+        }
     }
 }

src/JiShe.CollectBus.FreeRedisProvider/IFreeRedisProvider.cs

@@ -1,5 +1,4 @@
 using FreeRedis;
-using JiShe.CollectBus.Common.Models;
 
 namespace JiShe.CollectBus.FreeRedisProvider
 {

src/JiShe.CollectBus.FreeRedisProvider/Options/FreeRedisOptions.cs

@@ -13,11 +13,6 @@ namespace JiShe.CollectBus.FreeRedisProvider.Options
         /// </summary>
         public string? Configuration { get; set; }
 
-        /// <summary>
-        /// 最大连接数
-        /// </summary>
-        public string? MaxPoolSize { get; set; }
-
         /// <summary>
         /// 默认数据库
         /// </summary>

src/JiShe.CollectBus.Host/appsettings.json

@@ -41,7 +41,6 @@
   },
   "Redis": {
     "Configuration": "192.168.1.9:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
-    "MaxPoolSize": "50",
     "DefaultDB": "14",
     "HangfireDB": "15"
   },
@@ -86,8 +85,8 @@
     "BootstrapServers": "192.168.1.9:29092,192.168.1.9:39092,192.168.1.9:49092",
     "EnableFilter": true,
     "EnableAuthorization": false,
-    "SecurityProtocol": "SaslPlaintext",
+    "SecurityProtocol": "SASL_PLAINTEXT",
-    "SaslMechanism": "Plain",
+    "SaslMechanism": "PLAIN",
     "SaslUserName": "lixiao",
     "SaslPassword": "lixiao1980"
     //"Topic": {
@@ -130,7 +129,7 @@
     "OpenDebugMode": true,
     "UseTableSessionPoolByDefault": false
   },
-  "ServerTagName": "JiSheCollectBus3",
+  "ServerTagName": "JiSheCollectBus",
   "KafkaReplicationFactor": 3,
   "NumPartitions": 30,
   "Cassandra": {

src/JiShe.CollectBus.KafkaProducer/CollectBusKafkaModule.cs

@@ -16,15 +16,6 @@ namespace JiShe.CollectBus.Kafka
     {
         public override void ConfigureServices(ServiceConfigurationContext context)
         {
-            var configuration = context.Services.GetConfiguration();
-            var kafkaSection = configuration.GetSection("Kafka");
-            KafkaOptionConfig kafkaOptionConfig = new KafkaOptionConfig ();
-            kafkaSection.Bind(kafkaOptionConfig);
-            if (configuration["ServerTagName"] != null)
-            {
-                kafkaOptionConfig.ServerTagName = configuration["ServerTagName"]!;
-            }
-            context.Services.AddSingleton(kafkaOptionConfig);
             // 注册Producer
             context.Services.AddSingleton<IProducerService, ProducerService>();
             // 注册Consumer
@@ -34,12 +25,8 @@ namespace JiShe.CollectBus.Kafka
         public override void OnApplicationInitialization(ApplicationInitializationContext context)
         {
             var app = context.GetApplicationBuilder();
+            app.UseKafkaSubscribers(Assembly.Load("JiShe.CollectBus.Application"));
 
-            // 注册Subscriber
-            app.ApplicationServices.UseKafkaSubscribers();
-            
-            // 获取程序集
-            //app.UseKafkaSubscribers(Assembly.Load("JiShe.CollectBus.Application"));
         }
     }
 }

src/JiShe.CollectBus.KafkaProducer/Consumer/ConsumerService.cs

@@ -1,7 +1,16 @@
 using Confluent.Kafka;
 using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using JiShe.CollectBus.Kafka.Attributes;
+using Volo.Abp.DependencyInjection;
+using JiShe.CollectBus.Kafka.AdminClient;
+using static Confluent.Kafka.ConfigPropertyNames;
 using System.Collections.Concurrent;
+using System.Text.RegularExpressions;
+using NUglify.Html;
+using Serilog;
+using System;
 using System.Text;
 
 namespace JiShe.CollectBus.Kafka.Consumer
@@ -12,14 +21,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
         private readonly IConfiguration _configuration;
         private readonly ConcurrentDictionary<Type, (object Consumer, CancellationTokenSource CTS)>
             _consumerStore = new();
-        private readonly KafkaOptionConfig _kafkaOptionConfig;
         private class KafkaConsumer<TKey, TValue> where TKey : notnull where TValue : class { }
 
-        public ConsumerService(IConfiguration configuration, ILogger<ConsumerService> logger, KafkaOptionConfig kafkaOptionConfig)
+        public ConsumerService(IConfiguration configuration, ILogger<ConsumerService> logger)
         {
             _configuration = configuration;
             _logger = logger;
-            _kafkaOptionConfig = kafkaOptionConfig;
         }
 
         #region private 私有方法
@@ -42,9 +49,11 @@ namespace JiShe.CollectBus.Kafka.Consumer
 
         private ConsumerConfig BuildConsumerConfig(string? groupId = null)
         {
+            var enableAuth = bool.Parse(_configuration["Kafka:EnableAuthorization"]!);
+
             var config = new ConsumerConfig
             {
-                BootstrapServers = _kafkaOptionConfig.BootstrapServers,
+                BootstrapServers = _configuration["Kafka:BootstrapServers"],
                 GroupId = groupId ?? "default",
                 AutoOffsetReset = AutoOffsetReset.Earliest,
                 EnableAutoCommit = false, // 禁止AutoCommit
@@ -53,12 +62,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
                 FetchMaxBytes = 1024 * 1024 * 50 // 增加拉取大小（50MB）
             };
 
-            if (_kafkaOptionConfig.EnableAuthorization)
+            if (enableAuth)
             {
-                config.SecurityProtocol = _kafkaOptionConfig.SecurityProtocol;
+                config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
-                config.SaslMechanism = _kafkaOptionConfig.SaslMechanism;
+                config.SaslMechanism = SaslMechanism.Plain;
-                config.SaslUsername = _kafkaOptionConfig.SaslUserName;
+                config.SaslUsername = _configuration["Kafka:SaslUserName"];
-                config.SaslPassword = _kafkaOptionConfig.SaslPassword;
+                config.SaslPassword = _configuration["Kafka:SaslPassword"];
             }
 
             return config;
@@ -131,9 +140,9 @@ namespace JiShe.CollectBus.Kafka.Consumer
                             await Task.Delay(TimeSpan.FromSeconds(1),cts.Token);
                             continue;
                         }
-                        if (_kafkaOptionConfig.EnableFilter)
+                        if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
                         {
-                            var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) } };
+                            var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
                             // 检查 Header 是否符合条件
                             if (!headersFilter.Match(result.Message.Headers))
                             {
@@ -199,9 +208,9 @@ namespace JiShe.CollectBus.Kafka.Consumer
                             await Task.Delay(100, cts.Token);
                             continue;
                         }
-                        if (_kafkaOptionConfig.EnableFilter)
+                        if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
                         {
-                            var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) } };
+                            var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
                             // 检查 Header 是否符合条件
                             if (!headersFilter.Match(result.Message.Headers))
                             {
@@ -287,9 +296,9 @@ namespace JiShe.CollectBus.Kafka.Consumer
                                 }
                                 else if (result.Message.Value != null)
                                 {
-                                    if (_kafkaOptionConfig.EnableFilter)
+                                    if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
                                     {
-                                        var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) } };
+                                        var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
                                         // 检查 Header 是否符合条件
                                         if (!headersFilter.Match(result.Message.Headers))
                                         {
@@ -421,9 +430,9 @@ namespace JiShe.CollectBus.Kafka.Consumer
                                 }
                                 else if (result.Message.Value != null)
                                 {
-                                    if (_kafkaOptionConfig.EnableFilter)
+                                    if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
                                     {
-                                        var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) } };
+                                        var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
                                         // 检查 Header 是否符合条件
                                         if (!headersFilter.Match(result.Message.Headers))
                                         {

src/JiShe.CollectBus.KafkaProducer/IKafkaSubscribe.cs

@@ -6,12 +6,6 @@ using System.Threading.Tasks;
 
 namespace JiShe.CollectBus.Kafka
 {
-    /// <summary>
-    /// Kafka订阅者
-    /// <para>
-    /// 订阅者需要继承此接口并需要依赖注入，并使用<see cref="KafkaSubscribeAttribute"/>标记
-    /// </para>
-    /// </summary>
     public interface IKafkaSubscribe
     {
     }

src/JiShe.CollectBus.KafkaProducer/JiShe.CollectBus.Kafka.csproj

@@ -12,4 +12,8 @@
     <PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
   </ItemGroup>
 
+  <ItemGroup>
+    <ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
+  </ItemGroup>
+
 </Project>

src/JiShe.CollectBus.KafkaProducer/KafkaOptionConfig.cs

@@ -1,53 +0,0 @@
-using Confluent.Kafka;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Kafka
-{
-    public class KafkaOptionConfig
-    {
-        /// <summary>
-        /// kafka地址
-        /// </summary>
-        public string BootstrapServers { get; set; } = null!;
-
-        /// <summary>
-        /// 服务器标识
-        /// </summary>
-        public string ServerTagName { get; set; }= "KafkaFilterKey";
-
-        /// <summary>
-        /// 是否开启过滤器
-        /// </summary>
-        public bool EnableFilter { get; set; }= true;
-
-        /// <summary>
-        /// 是否开启认证
-        /// </summary>
-        public bool EnableAuthorization { get; set; } = false;
-
-        /// <summary>
-        /// 安全协议
-        /// </summary>
-        public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.SaslPlaintext;
-
-        /// <summary>
-        /// 认证方式
-        /// </summary>
-        public SaslMechanism SaslMechanism { get; set; }= SaslMechanism.Plain;
-
-        /// <summary>
-        /// 用户名
-        /// </summary>
-        public string? SaslUserName { get; set; }
-
-        /// <summary>
-        /// 密码
-        /// </summary>
-        public string? SaslPassword { get; set; }
-
-    }
-}

src/JiShe.CollectBus.KafkaProducer/KafkaSubcribesExtensions.cs

@@ -1,14 +1,24 @@
 using Confluent.Kafka;
-using JiShe.CollectBus.Kafka;
+using DeviceDetectorNET;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.Common.Helpers;
 using JiShe.CollectBus.Kafka.AdminClient;
 using JiShe.CollectBus.Kafka.Attributes;
 using JiShe.CollectBus.Kafka.Consumer;
 using Microsoft.AspNetCore.Builder;
-using Microsoft.AspNetCore.Mvc.ApplicationParts;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Primitives;
+using Newtonsoft.Json;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
 using System.Reflection;
+using System.Text;
+using System.Threading.Tasks;
+using static Confluent.Kafka.ConfigPropertyNames;
 
 namespace JiShe.CollectBus.Kafka
 {
@@ -19,55 +29,14 @@ namespace JiShe.CollectBus.Kafka
         /// </summary>
         /// <param name="app"></param>
         /// <param name="assembly"></param>
-        public static void UseKafkaSubscribers(this IServiceProvider provider)
-        {
-            var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
-
-            lifetime.ApplicationStarted.Register(() =>
-            {
-                var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>();
-                int threadCount = 0;
-                int topicCount = 0;
-                var assemblyPath = Path.GetDirectoryName(Assembly.GetEntryAssembly()?.Location);
-                if (string.IsNullOrWhiteSpace(assemblyPath))
-                {
-                    logger.LogInformation($"kafka订阅未能找到程序路径");
-                    return;
-                }
-                var dllFiles = Directory.GetFiles(assemblyPath, "*.dll");
-                foreach (var file in dllFiles)
-                {
-                    // 跳过已加载的程序集
-                    var assemblyName = AssemblyName.GetAssemblyName(file);
-                    var existingAssembly = AppDomain.CurrentDomain.GetAssemblies()
-                        .FirstOrDefault(a => a.GetName().FullName == assemblyName.FullName);
-                    var assembly = existingAssembly ?? Assembly.LoadFrom(file);
-                    // 实现IKafkaSubscribe接口
-                    var subscribeTypes = assembly.GetTypes().Where(type =>
-                                   typeof(IKafkaSubscribe).IsAssignableFrom(type) && 
-                                   !type.IsAbstract && !type.IsInterface).ToList(); ;
-                    if (subscribeTypes.Count == 0) 
-                        continue;
-                    foreach (var subscribeType in subscribeTypes)
-                    {
-                        var subscribes = provider.GetServices(subscribeType).ToList();
-                        subscribes.ForEach(subscribe =>
-                        {
-                            if (subscribe!=null)
-                            {
-                                Tuple<int, int> tuple = BuildKafkaSubscriber(subscribe, provider, logger);
-                                threadCount += tuple.Item1;
-                                topicCount += tuple.Item2;
-                            }
-                        });
-                    }
-                }
-                logger.LogInformation($"kafka订阅主题:{topicCount}数,共启动:{threadCount}线程");
-            });
-        }
-
         public static void UseKafkaSubscribers(this IApplicationBuilder app, Assembly assembly)
         {
+            var subscribeTypes = assembly.GetTypes()
+                .Where(t => typeof(IKafkaSubscribe).IsAssignableFrom(t))
+                .ToList();
+
+            if (subscribeTypes.Count == 0) return;
+
             var provider = app.ApplicationServices;
             var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
 
@@ -76,21 +45,16 @@ namespace JiShe.CollectBus.Kafka
                 var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>();
                 int threadCount = 0;
                 int topicCount = 0;
-                var subscribeTypes = assembly.GetTypes()
-                       .Where(t => typeof(IKafkaSubscribe).IsAssignableFrom(t))
-                       .ToList();
-
-                if (subscribeTypes.Count == 0) return;
                 foreach (var subscribeType in subscribeTypes)
                 {
                     var subscribes = provider.GetServices(subscribeType).ToList();
                     subscribes.ForEach(subscribe => { 
                     
-                        if (subscribe != null)
+                        if(subscribe is IKafkaSubscribe)
                         {
-                            Tuple<int, int> tuple = BuildKafkaSubscriber(subscribe, provider, logger);
+                            Tuple<int, int> tuple= BuildKafkaSubscriber(subscribe, provider, logger);
-                            threadCount += tuple.Item1;
+                            threadCount+= tuple.Item1;
-                            topicCount += tuple.Item2;
+                            topicCount+= tuple.Item2;
                         }
                     });
                 }

src/JiShe.CollectBus.KafkaProducer/Producer/ProducerService.cs

@@ -19,12 +19,11 @@ namespace JiShe.CollectBus.Kafka.Producer
         private readonly IConfiguration _configuration;
         private readonly ConcurrentDictionary<Type, object> _producerCache = new();
         private class KafkaProducer<TKey, TValue> where TKey : notnull where TValue : class { }
-        private readonly KafkaOptionConfig _kafkaOptionConfig;
+
-        public ProducerService(IConfiguration configuration,ILogger<ProducerService> logger, KafkaOptionConfig kafkaOptionConfig)
+        public ProducerService(IConfiguration configuration,ILogger<ProducerService> logger)
         {
             _configuration = configuration;
             _logger = logger;
-            _kafkaOptionConfig = kafkaOptionConfig;
         }
 
         #region private 私有方法
@@ -52,9 +51,11 @@ namespace JiShe.CollectBus.Kafka.Producer
         /// <returns></returns>
         private ProducerConfig BuildProducerConfig()
         {
+            var enableAuth = bool.Parse(_configuration["Kafka:EnableAuthorization"]!);
+
             var config = new ProducerConfig
             {
-                BootstrapServers = _kafkaOptionConfig.BootstrapServers,
+                BootstrapServers = _configuration["Kafka:BootstrapServers"],
                 AllowAutoCreateTopics = true,
                 QueueBufferingMaxKbytes = 2_097_151,  // 修改缓冲区最大为2GB，默认为1GB
                 CompressionType = CompressionType.Lz4, // 配置使用压缩算法LZ4，其他：gzip/snappy/zstd
@@ -65,12 +66,12 @@ namespace JiShe.CollectBus.Kafka.Producer
                 MessageTimeoutMs = 120000, // 消息发送超时时间为2分钟,设置值MessageTimeoutMs > LingerMs
             };
 
-            if (_kafkaOptionConfig.EnableAuthorization)
+            if (enableAuth)
             {
-                config.SecurityProtocol = _kafkaOptionConfig.SecurityProtocol;
+                config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
-                config.SaslMechanism = _kafkaOptionConfig.SaslMechanism;
+                config.SaslMechanism = SaslMechanism.Plain;
-                config.SaslUsername = _kafkaOptionConfig.SaslUserName;
+                config.SaslUsername = _configuration["Kafka:SaslUserName"];
-                config.SaslPassword = _kafkaOptionConfig.SaslPassword;
+                config.SaslPassword = _configuration["Kafka:SaslPassword"];
             }
 
             return config;
@@ -109,7 +110,7 @@ namespace JiShe.CollectBus.Kafka.Producer
                 Key = key,
                 Value = value,
                 Headers = new Headers{
-                   { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) }
+                   { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) }
                 }
             };
             await producer.ProduceAsync(topic, message);
@@ -130,7 +131,7 @@ namespace JiShe.CollectBus.Kafka.Producer
             {
                 Value = value,
                 Headers = new Headers{
-                   { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) }
+                   { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) }
                 }
             };
             await producer.ProduceAsync(topic, message);
@@ -154,7 +155,7 @@ namespace JiShe.CollectBus.Kafka.Producer
                 Key = key,
                 Value = value,
                 Headers = new Headers{
-                   { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) }
+                   { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) }
                 }
             };
             var typeKey = typeof(KafkaProducer<TKey, TValue>);
@@ -188,7 +189,7 @@ namespace JiShe.CollectBus.Kafka.Producer
             {
                 Value = value,
                 Headers = new Headers{
-                   { "route-key", Encoding.UTF8.GetBytes(_kafkaOptionConfig.ServerTagName) }
+                   { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) }
                 }
             };
             var typeKey = typeof(KafkaProducer<string, TValue>);

src/JiShe.CollectBus.Protocol.Contracts/Abstracts/BaseProtocolPlugin.cs

@@ -11,8 +11,9 @@ using Microsoft.Extensions.DependencyInjection;
 using JiShe.CollectBus.IotSystems.MessageReceiveds;
 using JiShe.CollectBus.IotSystems.Protocols;
 using MassTransit;
-using DotNetCore.CAP;
 using JiShe.CollectBus.Kafka.Producer;
+using JiShe.CollectBus.Common.Helpers;
+using DotNetCore.CAP;
 
 namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
 {

src/JiShe.CollectBus.Protocol.Contracts/JiShe.CollectBus.Protocol.Contracts.csproj

@@ -7,6 +7,7 @@
 	</PropertyGroup>
 
 	<ItemGroup>
+		<PackageReference Include="Confluent.Kafka" Version="2.9.0" />
 		<PackageReference Include="DotNetCore.CAP" Version="8.3.1" />
 		<PackageReference Include="MassTransit.Abstractions" Version="8.3.0" />
 		<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />