ASP.NET Core 3.x 并发限制

2020年03月30日 16:45:52   [来源:互联网]   [阅读:-]
字体:【

前言

Microsoft.AspNetCore.ConcurrencyLimiter AspNetCore3.0后增加的,用于传入的请求进行排队处理,避免线程池的不足.
我们日常开发中可能常做的给某web服务器配置连接数以及,请求队列大小,那么今天我们看看如何在通过中间件形式实现一个并发量以及队列长度限制.

Queue策略

添加Nuget

Install-Package Microsoft.AspNetCore.ConcurrencyLimiter

Copy public void ConfigureServices(IServiceCollection services)
{
services.AddQueuePolicy(options =>
{
//最大并发请求数
options.MaxConcurrentRequests = 2;
//请求队列长度限制
options.RequestQueueLimit = 1;
});
services.AddControllers();
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
//添加并发限制中间件
app.UseConcurrencyLimiter();
app.Run(async context =>
{
Task.Delay(100).Wait(); // 100ms sync-over-async

await context.Response.WriteAsync("Hello World!");
});
if (env.IsDevelopment())
{
app.UseDeveloperExceptionPage();
}

app.UseHttpsRedirection();

app.UseRouting();

app.UseAuthorization();

app.UseEndpoints(endpoints =>
{
endpoints.MapControllers();
});
}

通过上面简单的配置,我们就可以将他引入到我们的代码中,从而做并发量限制,以及队列的长度;那么问题来了,他是怎么实现的呢?

Copy public static IServiceCollection AddQueuePolicy(this IServiceCollection services, Action configure){ services.Configure(configure); services.AddSingleton();
return services;
}

QueuePolicy采用的是SemaphoreSlim信号量设计,SemaphoreSlim、Semaphore(信号量)支持并发多线程进入被保护代码,对象在初始化时会指定 最大任务数量,当线程请求访问资源,信号量递减,而当他们释放时,信号量计数又递增。

Copy ///  /// 构造方法(初始化Queue策略) ///  ///  public QueuePolicy(IOptions options) { _maxConcurrentRequests = options.Value.MaxConcurrentRequests; if (_maxConcurrentRequests <= 0) { throw new ArgumentException(nameof(_maxConcurrentRequests), "MaxConcurrentRequests must be a positive integer."); } _requestQueueLimit = options.Value.RequestQueueLimit; if (_requestQueueLimit < 0)
{
throw new ArgumentException(nameof(_requestQueueLimit), "The RequestQueueLimit cannot be a negative number.");
}
//使用SemaphoreSlim来限制任务最大个数
_serverSemaphore = new SemaphoreSlim(_maxConcurrentRequests);
}

ConcurrencyLimiterMiddleware中间件

Copy ///  /// Invokes the logic of the middleware. ///  /// The . /// A  that completes when the request leaves.
public async Task Invoke(HttpContext context)
{
var waitInQueueTask = _queuePolicy.TryEnterAsync();

// Make sure we only ever call GetResult once on the TryEnterAsync ValueTask b/c it resets.
bool result;

if (waitInQueueTask.IsCompleted)
{
ConcurrencyLimiterEventSource.Log.QueueSkipped();
result = waitInQueueTask.Result;
}
else
{
using (ConcurrencyLimiterEventSource.Log.QueueTimer())
{
result = await waitInQueueTask;
}
}

if (result)
{
try
{
await _next(context);
}
finally
{
_queuePolicy.OnExit();
}
}
else
{
ConcurrencyLimiterEventSource.Log.RequestRejected();
ConcurrencyLimiterLog.RequestRejectedQueueFull(_logger);
context.Response.StatusCode = StatusCodes.Status503ServiceUnavailable;
await _onRejected(context);
}
}

每次当我们请求的时候首先会调用_queuePolicy.TryEnterAsync(),进入该方法后先开启一个私有lock锁,再接着判断总请求量是否≥(请求队列限制的大小+最大并发请求数),如果当前数量超出了,那么我直接抛出,送你个503状态;

Copy if (result)
{
try
{
await _next(context);
}
finally
{
_queuePolicy.OnExit();
}
}
else
{
ConcurrencyLimiterEventSource.Log.RequestRejected();
ConcurrencyLimiterLog.RequestRejectedQueueFull(_logger);
context.Response.StatusCode = StatusCodes.Status503ServiceUnavailable;
await _onRejected(context);
}

问题来了,我这边如果说还没到你设置的大小呢,我这个请求没有给你服务器造不成压力,那么你给我处理一下吧.
await _serverSemaphore.WaitAsync();异步等待进入信号量,如果没有线程被授予对信号量的访问权限,则进入执行保护代码;否则此线程将在此处等待,直到信号量被释放为止

Copy lock (_totalRequestsLock)
{
if (TotalRequests >= _requestQueueLimit + _maxConcurrentRequests)
{
return false;
}
TotalRequests++;
}
//异步等待进入信号量,如果没有线程被授予对信号量的访问权限,则进入执行保护代码;否则此线程将在此处等待,直到信号量被释放为止
await _serverSemaphore.WaitAsync();
return true;
}

返回成功后那么中间件这边再进行处理,_queuePolicy.OnExit();通过该调用进行调用_serverSemaphore.Release();释放信号灯,再对总请求数递减

Stack策略

再来看看另一种方法,栈策略,他是怎么做的呢?一起来看看.再附加上如何使用的代码.

Copy public void ConfigureServices(IServiceCollection services)
{
services.AddStackPolicy(options =>
{
//最大并发请求数
options.MaxConcurrentRequests = 2;
//请求队列长度限制
options.RequestQueueLimit = 1;
});
services.AddControllers();
}

通过上面的配置,我们便可以对我们的应用程序执行出相应的策略.下面再来看看他是怎么实现的呢

Copy public static IServiceCollection AddStackPolicy(this IServiceCollection services, Action configure) { services.Configure(configure); services.AddSingleton();
return services;
}

可以看到这次是通过StackPolicy类做的策略.来一起来看看主要的方法

Copy ///  /// 构造方法(初始化参数) ///  ///  public StackPolicy(IOptions options) { //栈分配 _buffer = new List();
//队列大小
_maxQueueCapacity = options.Value.RequestQueueLimit;
//最大并发请求数
_maxConcurrentRequests = options.Value.MaxConcurrentRequests;
//剩余可用空间
_freeServerSpots = options.Value.MaxConcurrentRequests;
}

当我们通过中间件请求调用,_queuePolicy.TryEnterAsync()时,首先会判断我们是否还有访问请求次数,如果_freeServerSpots>0,那么则直接给我们返回true,让中间件直接去执行下一步,如果当前队列=我们设置的队列大小的话,那我们需要取消先前请求;每次取消都是先取消之前的保留后面的请求;

Copy public ValueTask TryEnterAsync() { lock (_bufferLock) { if (_freeServerSpots > 0) { _freeServerSpots--; return _trueTask; } // 如果队列满了,取消先前的请求 if (_queueLength == _maxQueueCapacity) { _hasReachedCapacity = true; _buffer[_head].Complete(false); _queueLength--; } var tcs = _cachedResettableTCS ??= new ResettableBooleanCompletionSource(this); _cachedResettableTCS = null; if (_hasReachedCapacity || _queueLength < _buffer.Count)
{
_buffer[_head] = tcs;
}
else
{
_buffer.Add(tcs);
}
_queueLength++;
// increment _head for next time
_head++;
if (_head == _maxQueueCapacity)
{
_head = 0;
}
return tcs.GetValueTask();
}
}

当我们请求后调用_queuePolicy.OnExit();出栈,再将请求长度递减

Copy public void OnExit()
{
lock (_bufferLock)
{
if (_queueLength == 0)
{
_freeServerSpots++;

if (_freeServerSpots > _maxConcurrentRequests)
{
_freeServerSpots--;
throw new InvalidOperationException("OnExit must only be called once per successful call to TryEnterAsync");
}

return;
}

// step backwards and launch a new task
if (_head == 0)
{
_head = _maxQueueCapacity - 1;
}
else
{
_head--;
}
//退出,出栈
_buffer[_head].Complete(true);
_queueLength--;
}
}

总结

基于栈结构的特点,在实际应用中,通常只会对栈执行以下两种操作:

  • 向栈中添加元素,此过程被称为"进栈"(入栈或压栈);
  • 从栈中提取出指定元素,此过程被称为"出栈"(或弹栈);

队列存储结构的实现有以下两种方式:

  • 顺序队列:在顺序表的基础上实现的队列结构;
  • 链队列:在链表的基础上实现的队列结构;

推荐阅读:苹果7p处理器


相关新闻
新闻焦点
在第一轮搜证当中,发现了很多的疑问,尤其是针对鬼花匠,她给两个死者都送了鸡汤,而且鸡汤没含有毒药,不得不怀疑凶手是不是她?黄明昊最怕鸡了,看到一[更多]
《大明风华》朱瞻基下线了。不过,朱亚文饰演的角色给观众们留下了深刻的印象。在剧中,朱瞻基年仅36岁就驾崩,被两个叔叔的精神打击与战场上旧病复发而[更多]
最近几年搞笑甜宠剧行业早已是一片红海!前不久上映的《灵剑山》深受观众的喜爱,好评不断,本以为近期找不出第二部,没想到啊,还真被小编发现了近期一部[更多]
最新一期《我们的歌》里,当费玉清和阿云嘎唱完之后,那英说:“年轻人有年轻人的表达方式,但是经过岁月洗礼的歌手唱歌,每个字都是如歌如泣!”的确,费[更多]
关于我们 | 联系我们 | 老版地图 | 网站地图 | 版权声明
版权所有:每日经济网未经授权禁止复制或建立镜像
相关作品的原创性、文中陈述文字以及内容数据庞杂本站无法一一核实,如果您发现本网站上有侵犯您的合法权益的内容,请联系我们,本网站将立即予以删除!
中国互联网违法和不良信息举报中心 网络警察报警岗亭