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);
}