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

using JiShe.CollectBus.FreeRedisProvider;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Volo.Abp.DependencyInjection;

namespace JiShe.CollectBus.Common.DeviceBalanceControl
{
    /// <summary>
    /// 设备组负载控制
    /// </summary>
    public class DeviceGroupBalanceControl
    {
        private static readonly object _syncRoot = new object();

        private static volatile CacheState _currentCache;

        /// <summary>
        /// 使用ConcurrentDictionary保证线程安全的设备分组映射
        /// </summary>
        private sealed class CacheState
        {
            public readonly ConcurrentDictionary<string, int> BalancedMapping;
            public readonly List<string>[] CachedGroups;

            public CacheState(int groupCount)
            {
                BalancedMapping = new ConcurrentDictionary<string, int>();
                CachedGroups = new List<string>[groupCount];
                for (int i = 0; i < groupCount; i++)
                {
                    CachedGroups[i] = new List<string>();
                }
            }
        }

        /// <summary>
        /// 初始化或增量更新缓存
        /// </summary>
        public static void InitializeCache(List<string> deviceList, int groupCount = 60)
        {
            lock (_syncRoot)
            {
                // 首次初始化
                if (_currentCache == null)
                {
                    var newCache = new CacheState(groupCount);
                    UpdateCacheWithDevices(newCache, deviceList, groupCount);
                    _currentCache = newCache;
                }
                // 后续增量更新
                else
                {
                    if (_currentCache.CachedGroups.Length != groupCount)
                        throw new ArgumentException("Group count cannot change after initial initialization");

                    var clonedCache = CloneExistingCache();
                    UpdateCacheWithDevices(clonedCache, deviceList, groupCount);
                    _currentCache = clonedCache;
                }
            }
        }

        /// <summary>
        /// 带锁的缓存克隆（写入时复制）
        /// </summary>
        private static CacheState CloneExistingCache()
        {
            var oldCache = _currentCache;
            var newCache = new CacheState(oldCache.CachedGroups.Length);

            // 复制已有映射
            foreach (var kvp in oldCache.BalancedMapping)
            {
                newCache.BalancedMapping.TryAdd(kvp.Key, kvp.Value);
            }

            // 复制分组数据
            for (int i = 0; i < oldCache.CachedGroups.Length; i++)
            {
                newCache.CachedGroups[i].AddRange(oldCache.CachedGroups[i]);
            }

            return newCache;
        }

        /// <summary>
        /// 更新设备到缓存
        /// </summary>
        private static void UpdateCacheWithDevices(CacheState cache, List<string> deviceList, int groupCount)
        {
            foreach (var deviceId in deviceList)
            {
                // 原子操作：如果设备不存在则计算分组
                cache.BalancedMapping.GetOrAdd(deviceId, id =>
                {
                    int groupId = GetGroupId(id, groupCount);
                    lock (cache.CachedGroups[groupId])
                    {
                        cache.CachedGroups[groupId].Add(id);
                    }
                    return groupId;
                });
            }
        }

        /// <summary>
        /// 并行处理泛型数据集（支持动态线程分配）
        /// </summary>
        /// <typeparam name="T">已经分组的设备信息</typeparam>
        /// <param name="items">部分或者全部的已经分组的设备集合</param>
        /// <param name="deviceIdSelector">从泛型对象提取deviceId</param>
        /// <param name="processor">处理委托（参数：当前对象，线程ID）</param>
        /// <param name="maxThreads">可选线程限制</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public static async Task ProcessGenericListAsync<T>(
            List<T> items, Func<T, string> deviceIdSelector, Action<T, int> processor, int? maxThreads = null)
        {
            var cache = _currentCache ?? throw new InvalidOperationException("缓存未初始化");

            // 创建分组任务队列
            var groupQueues = new ConcurrentQueue<T>[cache.CachedGroups.Length];
            for (int i = 0; i < groupQueues.Length; i++)
            {
                groupQueues[i] = new ConcurrentQueue<T>();
            }

            // 阶段1：分发数据到分组队列
            Parallel.ForEach(items, item =>
            {
                var deviceId = deviceIdSelector(item);
                if (cache.BalancedMapping.TryGetValue(deviceId, out int groupId))
                {
                    groupQueues[groupId].Enqueue(item);
                }
            });

            if ((maxThreads.HasValue && maxThreads.Value > cache.CachedGroups.Length) || maxThreads.HasValue == false)
            {
                maxThreads = cache.CachedGroups.Length;
            }

            // 阶段2：并行处理队列
            var options = new ParallelOptions
            {
                MaxDegreeOfParallelism = maxThreads.Value,
            };

            TimeSpan timeSpan = TimeSpan.FromMicroseconds(5);
            await Task.Run(() =>
            {
                Parallel.For(0, cache.CachedGroups.Length, options, async groupId =>
                {
                    var queue = groupQueues[groupId];
                    while (queue.TryDequeue(out T item))
                    {
                        await Task.Delay(timeSpan);
                        processor(item, Thread.CurrentThread.ManagedThreadId);
                    }
                });
            });
        }

        /// <summary>
        /// 并行处理所有分组设备（每个分组一个处理线程）
        /// </summary>
        //public static void ProcessAllGroups<T>(Action<List<T>> processAction) where T : DeviceGroupBasicModel
        //{
        //    var cache = _currentCache;
        //    if (cache == null)
        //        throw new InvalidOperationException("缓存未初始化");

        //    // 使用并行选项控制并发度
        //    var options = new ParallelOptions
        //    {
        //        MaxDegreeOfParallelism = cache.CachedGroups.Length // 严格匹配分组数量
        //    };

        //    Parallel.For(0, cache.CachedGroups.Length, options, groupId =>
        //    {
        //        // 获取当前分组的只读副本
        //        var groupDevices = GetGroupSnapshot(cache, groupId);

        //        processAction(groupDevices);

        //        //foreach (var deviceId in groupDevices)
        //        //{
        //        //    //执行处理操作
        //        //    processAction(deviceId);

        //        //    // 可添加取消检测
        //        //    // if (token.IsCancellationRequested) break;
        //        //}
        //    });
        //}

        ///// <summary>
        ///// 获取分组数据快照（线程安全）
        ///// </summary>
        //public static IReadOnlyList<string> GetGroupSnapshot(CacheState cache, int groupId)
        //{
        //    lock (cache.CachedGroups[groupId])
        //    {
        //        return cache.CachedGroups[groupId].ToList(); // 创建内存快照
        //    }
        //}

        /// <summary>
        /// 通过 deviceId 获取所在的分组集合
        /// </summary>
        public static List<string> GetGroup(string deviceId)
        {
            var cache = _currentCache;
            if (cache == null)
                throw new InvalidOperationException("缓存未初始化");

            return cache.CachedGroups[cache.BalancedMapping[deviceId]];
        }

        /// <summary>
        /// 通过 deviceId 获取分组Id
        /// </summary>
        public static int GetDeviceGroupId(string deviceId)
        {
            var cache = _currentCache;
            if (cache == null)
                throw new InvalidOperationException("缓存未初始化");

            return cache.BalancedMapping[deviceId];
        }


        /// <summary>
        /// 创建均衡映射表
        /// </summary>
        /// <param name="deviceList">数据集合</param>
        /// <param name="groupCount">分组数量</param>
        /// <param name="maxDeviation">允许的最大偏差百分比</param>
        /// <returns></returns>
        public static Dictionary<string, int> CreateBalancedMapping(List<string> deviceList, int groupCount, int maxDeviation = 5)
        {
            var mapping = new Dictionary<string, int>();
            int targetPerGroup = deviceList.Count / groupCount;
            int maxAllowed = (int)(targetPerGroup * (1 + maxDeviation / 100.0));

            // 初始化分组计数器
            int[] groupCounters = new int[groupCount];

            foreach (var deviceId in deviceList)
            {
                int preferredGroup = GetGroupId(deviceId, groupCount);

                // 如果首选分组未满，直接分配
                if (groupCounters[preferredGroup] < maxAllowed)
                {
                    mapping[deviceId] = preferredGroup;
                    groupCounters[preferredGroup]++;
                }
                else
                {
                    // 寻找当前最空闲的分组
                    int fallbackGroup = Array.IndexOf(groupCounters, groupCounters.Min());
                    mapping[deviceId] = fallbackGroup;
                    groupCounters[fallbackGroup]++;
                }
            }

            return mapping;
        }

        /// <summary>
        /// 分析分组分布
        /// </summary>
        /// <param name="deviceList"></param>
        /// <param name="groupCount"></param>
        /// <returns></returns>
        public static Dictionary<int, int> AnalyzeDistribution(List<string> deviceList, int groupCount)
        {
            Dictionary<int, int> distribution = new Dictionary<int, int>();
            foreach (var deviceId in deviceList)
            {
                int groupId = GetGroupId(deviceId, groupCount);
                distribution[groupId] = distribution.TryGetValue(groupId, out var count) ? count + 1 : 1;
            }
            return distribution;
        }

        /// <summary>
        /// 获取设备ID对应的分组ID
        /// </summary>
        /// <param name="deviceId"></param>
        /// <param name="groupCount"></param>
        /// <returns></returns>
        public static int GetGroupId(string deviceId, int groupCount)
        {
            int hash = Fnv1aHash(deviceId);
            // 双重取模确保分布均匀
            return (hash % groupCount + groupCount) % groupCount;
        }

        /// <summary>
        /// FNV-1a哈希算法
        /// </summary>
        /// <param name="input"></param>
        /// <returns></returns>
        public static int Fnv1aHash(string input)
        {
            const uint fnvPrime = 16777619;
            const uint fnvOffsetBasis = 2166136261;

            uint hash = fnvOffsetBasis;
            foreach (char c in input)
            {
                hash ^= (byte)c;
                hash *= fnvPrime;
            }
            return (int)hash;
        }

        /// <summary>
        /// 打印分组统计数据
        /// </summary>
        public static void PrintDistributionStats()
        {
            var cache = _currentCache;
            if (cache == null)
            {
                Console.WriteLine("缓存未初始化");
                return;
            }

            var stats = cache.CachedGroups
                .Select((group, idx) => new { GroupId = idx, Count = group.Count })
                .OrderBy(x => x.GroupId);

            Console.WriteLine("分组数据量统计:");
            foreach (var stat in stats)
            {
                Console.WriteLine($"Group {stat.GroupId}: {stat.Count} 条数据");
            }
        }

    }
}