ASP.NET Core 3.x 并發(fā)限制的實現(xiàn)代碼
更新時間：2019年11月13日 11:39:58 作者：馮輝
這篇文章主要介紹了ASP.NET Core 3.x 并發(fā)限制的實現(xiàn)代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
前言
Microsoft.AspNetCore.ConcurrencyLimiter AspNetCore3.0后增加的,用于傳入的請求進行排隊處理,避免線程池的不足.
我們?nèi)粘ｉ_發(fā)中可能常做的給某web服務器配置連接數(shù)以及,請求隊列大小,那么今天我們看看如何在通過中間件形式實現(xiàn)一個并發(fā)量以及隊列長度限制.
Queue策略
添加Nuget
Install-Package Microsoft.AspNetCore.ConcurrencyLimiter
public void ConfigureServices(IServiceCollection services)
{
services.AddQueuePolicy(options =>
{
//最大并發(fā)請求數(shù)
options.MaxConcurrentRequests = 2;
//請求隊列長度限制
options.RequestQueueLimit = 1;
});
services.AddControllers();
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
//添加并發(fā)限制中間件
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();
});
}
通過上面簡單的配置,我們就可以將他引入到我們的代碼中,從而做并發(fā)量限制,以及隊列的長度;那么問題來了,他是怎么實現(xiàn)的呢？
public static IServiceCollection AddQueuePolicy(this IServiceCollection services, Action<QueuePolicyOptions> configure)
{
services.Configure(configure);
services.AddSingleton<IQueuePolicy, QueuePolicy>();
return services;
}
QueuePolicy采用的是SemaphoreSlim信號量設計,SemaphoreSlim、Semaphore（信號量）支持并發(fā)多線程進入被保護代碼,對象在初始化時會指定 最大任務數(shù)量,當線程請求訪問資源,信號量遞減,而當他們釋放時,信號量計數(shù)又遞增。
/// <summary>
/// 構(gòu)造方法(初始化Queue策略)
/// </summary>
/// <param name="options"></param>
public QueuePolicy(IOptions<QueuePolicyOptions> 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來限制任務最大個數(shù)
_serverSemaphore = new SemaphoreSlim(_maxConcurrentRequests);
}
ConcurrencyLimiterMiddleware中間件
/// <summary>
/// Invokes the logic of the middleware.
/// </summary>
/// <param name="context">The <see cref="HttpContext"/>.</param>
/// <returns>A <see cref="Task"/> that completes when the request leaves.</returns>
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);
}
}
每次當我們請求的時候首先會調(diào)用_queuePolicy.TryEnterAsync(),進入該方法后先開啟一個私有l(wèi)ock鎖,再接著判斷總請求量是否≥(請求隊列限制的大小+最大并發(fā)請求數(shù)),如果當前數(shù)量超出了,那么我直接拋出,送你個503狀態(tài);
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();異步等待進入信號量，如果沒有線程被授予對信號量的訪問權(quán)限，則進入執(zhí)行保護代碼；否則此線程將在此處等待，直到信號量被釋放為止
lock (_totalRequestsLock)
{
if (TotalRequests >= _requestQueueLimit + _maxConcurrentRequests)
{
return false;
}
TotalRequests++;
}
//異步等待進入信號量，如果沒有線程被授予對信號量的訪問權(quán)限，則進入執(zhí)行保護代碼；否則此線程將在此處等待，直到信號量被釋放為止
await _serverSemaphore.WaitAsync();
return true;
}
返回成功后那么中間件這邊再進行處理,_queuePolicy.OnExit();通過該調(diào)用進行調(diào)用_serverSemaphore.Release();釋放信號燈,再對總請求數(shù)遞減
Stack策略
再來看看另一種方法,棧策略,他是怎么做的呢？一起來看看.再附加上如何使用的代碼.
public void ConfigureServices(IServiceCollection services)
{
services.AddStackPolicy(options =>
{
//最大并發(fā)請求數(shù)
options.MaxConcurrentRequests = 2;
//請求隊列長度限制
options.RequestQueueLimit = 1;
});
services.AddControllers();
}
通過上面的配置,我們便可以對我們的應用程序執(zhí)行出相應的策略.下面再來看看他是怎么實現(xiàn)的呢
public static IServiceCollection AddStackPolicy(this IServiceCollection services, Action<QueuePolicyOptions> configure)
{
services.Configure(configure);
services.AddSingleton<IQueuePolicy, StackPolicy>();
return services;
}
可以看到這次是通過StackPolicy類做的策略.來一起來看看主要的方法
/// <summary>
/// 構(gòu)造方法(初始化參數(shù))
/// </summary>
/// <param name="options"></param>
public StackPolicy(IOptions<QueuePolicyOptions> options)
{
//棧分配
_buffer = new List<ResettableBooleanCompletionSource>();
//隊列大小
_maxQueueCapacity = options.Value.RequestQueueLimit;
//最大并發(fā)請求數(shù)
_maxConcurrentRequests = options.Value.MaxConcurrentRequests;
//剩余可用空間
_freeServerSpots = options.Value.MaxConcurrentRequests;
}
當我們通過中間件請求調(diào)用,_queuePolicy.TryEnterAsync()時,首先會判斷我們是否還有訪問請求次數(shù),如果_freeServerSpots＞0,那么則直接給我們返回true,讓中間件直接去執(zhí)行下一步,如果當前隊列＝我們設置的隊列大小的話,那我們需要取消先前請求;每次取消都是先取消之前的保留后面的請求;
public ValueTask<bool> 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();
}
}
當我們請求后調(diào)用_queuePolicy.OnExit();出棧,再將請求長度遞減
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--;
}
}
總結(jié)
基于棧結(jié)構(gòu)的特點，在實際應用中，通常只會對棧執(zhí)行以下兩種操作：
- 向棧中添加元素，此過程被稱為"進棧"（入棧或壓棧）；
- 從棧中提取出指定元素，此過程被稱為"出棧"（或彈棧）；
隊列存儲結(jié)構(gòu)的實現(xiàn)有以下兩種方式：
- 順序隊列：在順序表的基礎上實現(xiàn)的隊列結(jié)構(gòu)；
- 鏈隊列：在鏈表的基礎上實現(xiàn)的隊列結(jié)構(gòu)；
以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
您可能感興趣的文章:
相關文章
解析ASP.NET?Core中Options模式的使用及其源碼
這篇文章主要介紹了ASP.NET?Core中Options模式的使用及其源碼解析,在ASP.NET Core中引入了Options這一使用配置方式，其主要是為了解決依賴注入時需要傳遞指定數(shù)據(jù)問題（不是自行獲取，而是能集中配置），需要的朋友可以參考下2022-03-03
訂單號在購物過程中起到了很好的識別作用，更方便的有利于工作人員識別商品，本文介紹利用asp.net實現(xiàn)生成訂單號2012-12-12
這篇文章主要介紹了ASP.NET文件處理如何操作，需要的朋友可以參考下2015-09-09
ASP.NET MVC 3仿Server.Transfer效果的實現(xiàn)方法
這篇文章主要介紹了ASP.NET MVC 3仿Server.Transfer效果的實現(xiàn)方法，需要的朋友可以參考下2015-10-10
這篇文章主要給大家介紹了關于如何給asp.net core寫個中間件記錄接口耗時的相關資料，文中通過示例代碼介紹的非常詳細，對大家學習或者使用asp.net core具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2019-09-09
ASP.NET Core利用UrlFirewall對請求進行過濾的方法示例
這篇文章主要給大家介紹了關于ASP.NET Core利用UrlFirewall對請求進行過濾的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-05-05
之前就了解過混淆加密工具xenocode，下面記錄一下xenocode2010代碼混淆的操作步驟，大家參考使用吧2014-01-01
asp.net 2.0中的App_GlobalResources可以用來解決本地化的問題，程序會根據(jù)瀏覽器的語言首選項自動判斷顯示出本地化的界面。2008-09-09
在.NET平臺中操作生成PDF的類庫有很多如常見的有iTextSharp、PDFsharp、Aspose.PDF等,本文主要介紹的是如何使用QuestPDF實現(xiàn)發(fā)票PDF文檔生成功能,需要的可以了解下2024-01-01
今年有個系統(tǒng)的部分EXCEL的操作也讓我做，順便結(jié)合之前操作EXCEL的經(jīng)驗作一下總結(jié)，可能也算不上什么，對于絕大多數(shù)來說也沒什么技術(shù)含量，網(wǎng)上一搜一大把，但我想還是有必要總結(jié)一下2011-02-02