带有权重的服务器SLB的实现

Posted 2020-09-27 Serviceboy

1）参考了网络上的算法，但是那个算法仅仅是用于展示“权重轮循”的意图，在真正的网络下，因为是并行的，所以不可能单纯一个简单的循环可以解决问题。

2）用lock的话性能显然有损失。

3）想了一阵，结合CAS和volatile等细粒度的锁的方式，一个真正可以用软件描述SLB带有权重的算法大概是这个样子（如下）：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace WBasedRobin
{
    /// <summary>
    /// 用于计算WeightRobin的数据结构
    /// </summary>
    public class WeightedRobin
    {
        private readonly int _weight;
        private int _count;
        /// <summary>
        /// 命中次数（累加）
        /// </summary>
        public int ChoosenCount
        {
            get
            {
                return ++_count;
            }
        }
        /// <summary>
        /// 权重
        /// </summary>
        public int Weight
        {
            get
            {
                return _weight;
            }
        }
        /// <summary>
        /// 输出当前的权重
        /// </summary>
        public override string ToString()
        {
            return "Weight:" + Weight.ToString() + "\tCount:" + _count.ToString();
        }
        /// <summary>
        /// 初始化每一个Server的内部值
        /// </summary>
        public WeightedRobin(int weight, int count = 0)
        {
            _weight = weight;
            _count = count;
        }
    }

    public class WeightRobinRule
    {
        private List<WeightedRobin> _servers = null;

        private volatile int _index = -1;
        private volatile int _currentWeight = 0;
        private volatile bool _isServerChanging = false;

        private volatile int _maxWeight = 0;
        private volatile int _gcdWeight = 0;

        private int GetMaxWeight(IEnumerable<WeightedRobin> weights)
        {
            return weights.Max(w => w.Weight);
        }

        private int GetGCDWeight(int big, int small)
        {
            if (big < small)
            {
                big ^= small;
                small ^= big;
                big ^= small;
            }

            if (big % small == 0)
            {
                return small;
            }
            return GetGCDWeight(small, big % small);
        }

        private int GetTotalGCD()
        {
            int gcd = GetGCDWeight(_servers[0].Weight, _servers[1].Weight);

            for (int i = 2; i < _servers.Count; ++i)
            {
                gcd = GetGCDWeight(gcd, _servers[i].Weight);
            }

            return gcd;
        }

        /// <summary>
        /// 初始化权重服务器，至少2台服务器。
        /// </summary>
        public WeightRobinRule(int totalServers = 2)
        {
            Random r = new Random();
            _servers = new List<WeightedRobin>(totalServers);

            for (int i = 0; i < totalServers; i++)
            {
                _servers.Add(new WeightedRobin(r.Next(2, totalServers+1),0));
            }
            _maxWeight = GetMaxWeight(_servers);
            _gcdWeight = GetTotalGCD();
        }

        public void DoRolling()
        {
            int copyIndex = 0;
            int copyIndexNext = 0;
            int copycw = 0;

        //当服务器数量发生变化的时候，锁住该服务直到完毕。
        reloop:   while (_isServerChanging) ;

            for (;;)
            {
                //拷贝本地的index，用做同步
                copyIndex = _index;
                //计算轮询的时候下一个的值
                copyIndexNext = (copyIndex + 1) % _servers.Count;
                //同步作用
                copycw = _currentWeight;

                //假定轮询后的Next=0，说明完成一轮轮询，权重减去最大公约数
                if (copyIndexNext == 0)
                {
                    copycw -= _gcdWeight;

                    //如果权重已经扣完，重新从大的开始
                    if (copycw <= 0)
                    {
                        copycw = _maxWeight;
                    }
                }

                //如果copyIndex和_index相同，说明是同一个线程抢到的，那么直接用本地的替换index进行替换
                if (Interlocked.CompareExchange(ref _index, copyIndexNext, copyIndex) == copyIndex)
                {
                    _currentWeight = copycw;

                    try
                    {
                        //如果轮询的权重大于等于本地权重，选中它即可。
                        if (_servers[copyIndexNext].Weight >= copycw)
                        {
                          int t =  _servers[copyIndexNext].ChoosenCount;
                            break;
                        }
                    }
                    //如果是Index溢出，那么说明服务器数量肯定发生变化了，所以跳过此次轮询，等下一轮，不处理。
                    catch (IndexOutOfRangeException)
                    {
                        goto reloop;
                    }

                }
            }
        }
        /// <summary>
        /// 移除指定的服务器
        /// </summary>
        public WeightedRobin RemoveByIndex(int index)
        {
            _isServerChanging = true;
            var removedServer = _servers[index];
            _servers.RemoveAt(index);
            _gcdWeight = GetTotalGCD();
            _maxWeight = GetMaxWeight(_servers);
            _isServerChanging = false;
            return removedServer;
        }
        /// <summary>
        /// 增加新的服务器
        /// </summary>
        public void AddNewServer(int weight)
        {
            _isServerChanging = true;
            _servers.Add(new WeightedRobin(weight, 0));
            _gcdWeight = GetTotalGCD();
            _maxWeight = GetMaxWeight(_servers);
            _isServerChanging = false;
        }
        /// <summary>
        /// 格式化输出结果
        /// </summary>
        public override string ToString()
        {
            StringBuilder sbu = new StringBuilder(10);

            foreach (WeightedRobin wr in _servers)
            {
                sbu.AppendLine(wr.ToString() + Environment.NewLine);
            }
            return sbu.ToString();
        }
    }
}

调用测试代码如下：

using System;
using System.Threading;
using System.Threading.Tasks;

namespace WBasedRobin
{
    class Program
    {
        static Random r = new Random();

        static void Rounding(WeightRobinRule wr)
        {
            wr.DoRolling();
        }
        static void Main(string[] args)
        {
            WeightRobinRule wr = new WeightRobinRule(5);

            Timer t = new Timer((j) => { var removedS = wr.RemoveByIndex(0); Console.WriteLine("移除了服务器："+removedS);  }, null, 2050, Timeout.Infinite);

             t = new Timer((o) => { wr.AddNewServer(6); Console.WriteLine("新增加服务器了。"); }, null, 3000, Timeout.Infinite);

            Parallel.For(1, 1001, (num) => 
            {
                Thread.Sleep(r.Next(100, 500));
                Rounding(wr);
            });
            Console.WriteLine(wr);
            Console.ReadLine();
        }
    }
}