如何在Web API中控制单用户针对特定Action的活跃请求数？

我正好做过类似的需求，给你分享几个针对异步Web API的用户级活跃请求限流方案，核心思路是跟踪每个用户当前未完成的异步请求数量，达到阈值时拒绝新请求，下面以ASP.NET Core为例（其他框架逻辑类似）：

一、单节点部署场景：内存缓存+Action过滤器 #

如果你的API是单服务器部署，用内存缓存就能搞定，实现起来简单高效：

1. 实现限流过滤器 #

创建一个异步Action过滤器，在请求执行前后维护用户的活跃请求计数，注意用原子操作避免并发问题：

public class ActiveRequestLimitFilter : IAsyncActionFilter
{
    private readonly IMemoryCache _cache;
    private readonly int _maxActiveRequests;

    public ActiveRequestLimitFilter(IMemoryCache cache, IConfiguration configuration)
    {
        _cache = cache;
        // 从配置文件读取单用户最大活跃请求数
        _maxActiveRequests = configuration.GetValue<int>("ActiveRequestLimit:MaxPerUser");
    }

    public async Task OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)
    {
        // 获取用户唯一标识（根据你的认证方式调整，比如JWT里的用户ID）
        var userId = context.HttpContext.User.FindFirstValue(ClaimTypes.NameIdentifier);
        if (string.IsNullOrEmpty(userId))
        {
            // 未认证用户可根据需求选择放行或限制，这里直接放行
            await next();
            return;
        }

        var cacheKey = $"ActiveRequests:{userId}";
        
        // 先获取当前计数，初始为0
        var currentCount = _cache.GetOrCreate(cacheKey, entry => 0);
        if (currentCount >= _maxActiveRequests)
        {
            // 超过阈值，返回429
            context.Result = new StatusCodeResult(StatusCodes.Status429TooManyRequests);
            return;
        }

        // 原子递增计数，避免并发下的计数错误
        _cache.Set(cacheKey, Interlocked.Increment(ref currentCount));

        try
        {
            // 执行Action
            await next();
        }
        finally
        {
            // 不管请求成功/失败/超时，都递减计数
            var count = _cache.Get<int>(cacheKey);
            if (count > 0)
            {
                _cache.Set(cacheKey, Interlocked.Decrement(ref count));
            }
            else
            {
                // 计数为0时移除缓存键，节省内存
                _cache.Remove(cacheKey);
            }
        }
    }
}

2. 注册过滤器与配置 #

在Program.cs中注册过滤器，并在appsettings.json配置阈值：

// 注册过滤器到全局
builder.Services.AddControllers(options =>
{
    options.Filters.Add<ActiveRequestLimitFilter>();
});

// appsettings.json
"ActiveRequestLimit": {
    "MaxPerUser": 3 // 单用户最多同时3个未完成请求
}

二、多节点部署场景：分布式缓存（Redis）+过滤器 #

如果是多服务器集群部署，内存缓存无法跨节点共享计数，这时候用Redis的原子命令来维护计数：

1. 实现分布式限流过滤器 #

public class DistributedActiveRequestLimitFilter : IAsyncActionFilter
{
    private readonly IDistributedCache _distributedCache;
    private readonly int _maxActiveRequests;
    // 设置缓存过期时间，防止因异常导致计数永久残留
    private readonly TimeSpan _cacheExpiration = TimeSpan.FromMinutes(10);

    public DistributedActiveRequestLimitFilter(IDistributedCache distributedCache, IConfiguration configuration)
    {
        _distributedCache = distributedCache;
        _maxActiveRequests = configuration.GetValue<int>("ActiveRequestLimit:MaxPerUser");
    }

    public async Task OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)
    {
        var userId = context.HttpContext.User.FindFirstValue(ClaimTypes.NameIdentifier);
        if (string.IsNullOrEmpty(userId))
        {
            await next();
            return;
        }

        var cacheKey = $"ActiveRequests:{userId}";
        
        // Redis原子递增计数
        var currentCount = await _distributedCache.StringIncrementAsync(cacheKey);
        
        // 首次递增时设置过期时间，避免每次覆盖
        if (currentCount == 1)
        {
            await _distributedCache.SetStringAsync(cacheKey, currentCount.ToString(), new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = _cacheExpiration
            });
        }

        if (currentCount > _maxActiveRequests)
        {
            // 超过阈值，把刚才递增的计数减回去
            await _distributedCache.StringDecrementAsync(cacheKey);
            context.Result = new StatusCodeResult(StatusCodes.Status429TooManyRequests);
            return;
        }

        try
        {
            await next();
        }
        finally
        {
            // 递减计数，计数为0时移除缓存键
            var count = await _distributedCache.StringDecrementAsync(cacheKey);
            if (count <= 0)
            {
                await _distributedCache.RemoveAsync(cacheKey);
            }
        }
    }
}

2. 注册分布式缓存与过滤器 #

// 注册Redis分布式缓存
builder.Services.AddStackExchangeRedisCache(options =>
{
    options.Configuration = builder.Configuration.GetConnectionString("Redis");
});

// 注册分布式限流过滤器
builder.Services.AddControllers(options =>
{
    options.Filters.Add<DistributedActiveRequestLimitFilter>();
});

三、进阶优化：针对特定Action限流 #

如果不需要全局限流，只想给耗时的异步Action单独设置限制，可以用特性标记：

1. 创建自定义特性：

[AttributeUsage(AttributeTargets.Method | AttributeTargets.Class)]
public class LimitActiveRequestsAttribute : Attribute
{
}

2. 修改过滤器，只对带特性的Action生效：

public async Task OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next)
{
    // 检查当前Action是否带有限流特性
    var hasLimitAttribute = context.ActionDescriptor.EndpointMetadata.OfType<LimitActiveRequestsAttribute>().Any();
    if (!hasLimitAttribute)
    {
        await next();
        return;
    }

    // 后续限流逻辑不变...
}

3. 在目标Action上标记：

[HttpPost("long-running-task")]
[LimitActiveRequests]
public async Task<IActionResult> LongRunningAsyncTask()
{
    // 耗时的异步操作
    await Task.Delay(TimeSpan.FromMinutes(3));
    return Ok("任务完成");
}

关键注意事项 #

• 用户标识准确性：一定要用可靠的用户唯一标识（如认证后的用户ID），不要用IP，避免多用户共享IP导致误限流。
• 异常与超时处理：finally块会确保请求无论成功、失败还是超时，计数都会被正确递减，避免计数残留。
• 缓存降级：如果分布式缓存（如Redis）不可用，可以添加try/catch块，临时放行请求或返回503服务不可用，避免影响业务。

内容的提问来源于stack exchange，提问作者user3407500