public class LimitingFactory { /// /// 创建限流服务对象 /// /// 限流模型 /// 最大QPS /// 最大可用票据数 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); } } } /// /// 限流模式 /// public enum LimitingType { TokenBucket,//令牌桶模式 LeakageBucket//漏桶模式 } public class LimitedQueue : Queue { 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; } }
调用方法:
var service = LimitingFactory.Build(LimitingType.TokenBucket, 500, 200); while (true) { var result = service.Request(); //如果返回true,说明可以进行业务处理,否则需要继续等待 if (result) { //业务处理...... } else Thread.Sleep(1); }
class LeakageBucketLimitingService: ILimitingService { private LimitedQueue 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(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()); } } }
调用方法:
var service = LimitingFactory.Build(LimitingType.LeakageBucket, 500, 200); while (true) { var result = service.Request(); //如果返回true,说明可以进行业务处理,否则需要继续等待 if (result) { //业务处理...... } else Thread.Sleep(1); }