ASP.NETCore3.x并发限制的实现代码

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

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

   /// 

    ///   构造方法(初始化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);
    }

    /// 

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

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

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

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

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

    /// 

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

  public ValueTask TryEnterAsync()
    {
      lock (_bufferLock)
      {
        if (_freeServerSpots > 0)
        {
          _freeServerSpots--;
          return _trueTask;
        }
        // 如果队列满了，取消先前的请求
        if (_queueLength == _maxQueueCapacity)
        {
          _hasReachedCapacity = true;
          _buffer[_head].Complete(false);
          _queueLength--;
        }
        var tcs = _cachedResettableTCS &＃63;&＃63;= 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();
      }
    }

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