基于.net的分布式系统限流组件（限流算法：令牌算法和漏斗算法）

转载连接：https://www.cnblogs.com/vveiliang/p/9049393.html

 

一、令牌桶算法
令牌桶算法是比较常见的限流算法之一，大概描述以下：
1）、全部的请求在处理以前都须要拿到一个可用的令牌才会被处理；
2）、根据限流大小，设置按照必定的速率往桶里添加令牌；
3）、桶设置最大的放置令牌限制，当桶满时、新添加的令牌就被丢弃活着拒绝；
4）、请求达到后首先要获取令牌桶中的令牌，拿着令牌才能够进行其余的业务逻辑，处理完业务逻辑以后，将令牌直接删除；
5）、令牌桶有最低限额，当桶中的令牌达到最低限额的时候，请求处理完以后将不会删除令牌，以此保证足够的限流；

 

 

 

 

二、漏桶算法

漏桶算法其实很简单，能够粗略的认为就是注水漏水过程，往桶中以必定速率流出水，以任意速率流入水，当水超过桶流量则丢弃，由于桶容量是不变的，保证了总体的速率

 

 

C# 代码实现：

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace 限流算法
{
    class Program
    {
        static void Main(string[] args)
        {
            var service = LimitingFactory.Build(LimitingType.LeakageBucket, 500, 200);

            while (true)
            {
                var result = service.Request();
                //若是返回true，说明能够进行业务处理，不然须要继续等待
                if (result)
                {
                    //业务处理......
                }
                else
                    Thread.Sleep(1);
            }

            Console.WriteLine("Hello World!");
        }
    }

    public interface ILimitingService : IDisposable
    {
        /// <summary>
        /// 申请流量处理
        /// </summary>
        /// <returns>true：获取成功，false：获取失败</returns>
        bool Request();

    }

    public class LimitingFactory
    {
        /// <summary>
        /// 建立限流服务对象
        /// </summary>
        /// <param name="limitingType">限流模型</param>
        /// <param name="maxQPS">最大QPS</param>
        /// <param name="limitSize">最大可用票据数</param>
        public static ILimitingService Build(LimitingType limitingType = LimitingType.TokenBucket, int maxQPS = 100, int limitSize = 100)
        {
            switch (limitingType)
            {
                case LimitingType.TokenBucket:
                default:
                    return new TokenBucketLimitingService(maxQPS, limitSize);
                case LimitingType.LeakageBucket:
                    return new LeakageBucketLimitingService(maxQPS, limitSize);
            }
        }
    }



    /// <summary>
    /// 限流模式
    /// </summary>
    public enum LimitingType
    {
        TokenBucket,//令牌桶模式
        LeakageBucket//漏桶模式
    }



    public class LimitedQueue<T> : Queue<T>
    {
        private int limit = 0;
        public const string QueueFulled = "TTP-StreamLimiting-1001";

        public int Limit
        {
            get { return limit; }
            set { limit = value; }
        }

        public LimitedQueue()
             : this(0)
        { }

        public LimitedQueue(int limit)
             : base(limit)
        {
            this.Limit = limit;
        }

        public new bool Enqueue(T item)
        {
            if (limit > 0 && this.Count >= this.Limit)
            {
                return false;
            }
            base.Enqueue(item);
            return true;
        }
    }

    class TokenBucketLimitingService : ILimitingService
    {
        private LimitedQueue<object> limitedQueue = null;
        private CancellationTokenSource cancelToken;
        private Task task = null;
        private int maxTPS;
        private int limitSize;
        private object lckObj = new object();
        public TokenBucketLimitingService(int maxTPS, int limitSize)
        {
            this.limitSize = limitSize;
            this.maxTPS = maxTPS;

            if (this.limitSize <= 0)
                this.limitSize = 100;
            if (this.maxTPS <= 0)
                this.maxTPS = 1;

            limitedQueue = new LimitedQueue<object>(limitSize);
            for (int i = 0; i < limitSize; i++)
            {
                limitedQueue.Enqueue(new object());
            }
            cancelToken = new CancellationTokenSource();
            task = Task.Factory.StartNew(new Action(TokenProcess), cancelToken.Token);
        }

        /// <summary>
        /// 定时消息令牌
        /// </summary>
        private void TokenProcess()
        {
            int sleep = 1000 / maxTPS;
            if (sleep == 0)
                sleep = 1;

            DateTime start = DateTime.Now;
            while (cancelToken.Token.IsCancellationRequested == false)
            {
                try
                {
                    lock (lckObj)
                    {
                        limitedQueue.Enqueue(new object());
                    }
                }
                catch
                {
                }
                finally
                {
                    if (DateTime.Now - start < TimeSpan.FromMilliseconds(sleep))
                    {
                        int newSleep = sleep - (int)(DateTime.Now - start).TotalMilliseconds;
                        if (newSleep > 1)
                            Thread.Sleep(newSleep - 1); //作一下时间上的补偿
                    }
                    start = DateTime.Now;
                }
            }
        }

        public void Dispose()
        {
            cancelToken.Cancel();
        }

        /// <summary>
        /// 请求令牌
        /// </summary>
        /// <returns>true：获取成功，false：获取失败</returns>
        public bool Request()
        {
            if (limitedQueue.Count <= 0)
                return false;
            lock (lckObj)
            {
                if (limitedQueue.Count <= 0)
                    return false;

                object data = limitedQueue.Dequeue();
                if (data == null)
                    return false;
            }

            return true;
        }
    }

    class LeakageBucketLimitingService : ILimitingService
    {
        private LimitedQueue<object> limitedQueue = null;
        private CancellationTokenSource cancelToken;
        private Task task = null;
        private int maxTPS;
        private int limitSize;
        private object lckObj = new object();
        public LeakageBucketLimitingService(int maxTPS, int limitSize)
        {
            this.limitSize = limitSize;
            this.maxTPS = maxTPS;

            if (this.limitSize <= 0)
                this.limitSize = 100;
            if (this.maxTPS <= 0)
                this.maxTPS = 1;

            limitedQueue = new LimitedQueue<object>(limitSize);
            cancelToken = new CancellationTokenSource();
            task = Task.Factory.StartNew(new Action(TokenProcess), cancelToken.Token);
        }

        private void TokenProcess()
        {
            int sleep = 1000 / maxTPS;
            if (sleep == 0)
                sleep = 1;

            DateTime start = DateTime.Now;
            while (cancelToken.Token.IsCancellationRequested == false)
            {
                try
                {

                    if (limitedQueue.Count > 0)
                    {
                        lock (lckObj)
                        {
                            if (limitedQueue.Count > 0)
                                limitedQueue.Dequeue();
                        }
                    }
                }
                catch
                {
                }
                finally
                {
                    if (DateTime.Now - start < TimeSpan.FromMilliseconds(sleep))
                    {
                        int newSleep = sleep - (int)(DateTime.Now - start).TotalMilliseconds;
                        if (newSleep > 1)
                            Thread.Sleep(newSleep - 1); //作一下时间上的补偿
                    }
                    start = DateTime.Now;
                }
            }
        }

        public void Dispose()
        {
            cancelToken.Cancel();
        }

        public bool Request()
        {
            if (limitedQueue.Count >= limitSize)
                return false;
            lock (lckObj)
            {
                if (limitedQueue.Count >= limitSize)
                    return false;

                return limitedQueue.Enqueue(new object());
            }
        }
    }
}

 

漏桶算法 VS 令牌桶算法

漏桶算法和令牌桶算法最明显的区别是令牌桶算法容许流量必定程度的突发。由于默认的令牌桶算法，取走token是不须要耗费时间的，也就是说，假设桶内有100个token时，那么能够瞬间容许100个请求经过。

令牌桶算法因为实现简单，且容许某些流量的突发，对用户友好，因此被业界采用地较多。固然咱们须要具体状况具体分析，只有最合适的算法，没有最优的算法。