重新整理 .net core 周边阅读篇————AspNetCoreRateLimit[一]

Posted 2021-09-24 程序员其实就是一个写文档的工作,代码只是文档的一部分,一切皆

前言

整理了一下.net core 一些常见的库的源码阅读，共32个库，记100余篇。

以下只是个人的源码阅读，如有错误或者思路不正确，望请指点。

正文

github 地址为:

https://github.com/stefanprodan/AspNetCoreRateLimit

一般个人习惯先阅读readme的简介。

上面大概翻译是：

AspNetCoreRateLimit 是ASP.NET Core 访问速率限制的解决方案，设计基于ip地址和客户端id用于控制用于web api和 mvc app的客户端访问速率。

这个包包含了IpRateLimitMiddleware and a ClientRateLimitMiddleware两个中间件，用这两个中间件你根据不同的场景能设置几种不同的限制，

比如限制一个客户端或者一个ip在几秒或者15分钟内访问最大限制。您可以定义这些限制来处理对某个API的所有请求，也可以将这些限制限定在指定范围的每个API URL或HTTP请求路径上。

上面说了这么多就是用来限流的，针对客户端id和ip进行限流。

因为一般用的是ip限流，看下ip限制怎么使用的，毕竟主要还是拿来用的嘛。

本来还想根据文档先写个小demo，然后发现官方已经写了demo。

直接看demo(只看ip限制部分的)。

先来看下ConfigureServices：

services.Configure<IpRateLimitOptions>(Configuration.GetSection("IpRateLimiting"));
services.Configure<IpRateLimitPolicies>(Configuration.GetSection("IpRateLimitPolicies"));
services.AddSingleton<IRateLimitConfiguration, RateLimitConfiguration>();
services.AddInMemoryRateLimiting();

从上面看，那么配置ip 限制的有两个配置，一个配置是IpRateLimitOptions，另外一个配置是IpRateLimitPolicies。

那么为什么要设计成两个配置呢？一个配置不是更香吗？

官方设计理念是这样的:

https://github.com/stefanprodan/AspNetCoreRateLimit/wiki/IpRateLimitMiddleware#defining-rate-limit-rules

IpRateLimiting Configuration and general rules appsettings.json

IpRateLimitPolicies Override general rules for specific IPs appsettings.json

原来IpRateLimiting 是限制普遍的ip，而IpRateLimitPolicies 是限制一些特殊的ip。

比如说有些api对内又对外的，普遍的ip对外限制是1分钟300次，如果有个大客户特殊需求且固定ip的，需要限制是1分钟是10000次的，那么就可以这样特殊处理，而不用另外写code来维护，成本问题。

故而我们写中间件组件的时候也可以参考这个来做，特殊的怎么处理，普遍的怎么处理，当然也不能盲目的设计。

然后看：AddInMemoryRateLimiting

public static IServiceCollection AddInMemoryRateLimiting(this IServiceCollection services)
{
	services.AddSingleton<IIpPolicyStore, MemoryCacheIpPolicyStore>();
	services.AddSingleton<IClientPolicyStore, MemoryCacheClientPolicyStore>();
	services.AddSingleton<IRateLimitCounterStore, MemoryCacheRateLimitCounterStore>();
	services.AddSingleton<IProcessingStrategy, AsyncKeyLockProcessingStrategy>();
	return services;
}

里面注入了MemoryCacheIpPolicyStore、MemoryCacheClientPolicyStore、MemoryCacheRateLimitCounterStore、AsyncKeyLockProcessingStrategy。

分别看下这几个东西。

MemoryCacheIpPolicyStore:

public class MemoryCacheIpPolicyStore : MemoryCacheRateLimitStore<IpRateLimitPolicies>, IIpPolicyStore
{
	private readonly IpRateLimitOptions _options;
	private readonly IpRateLimitPolicies _policies;

	public MemoryCacheIpPolicyStore(
		IMemoryCache cache,
		IOptions<IpRateLimitOptions> options = null,
		IOptions<IpRateLimitPolicies> policies = null) : base(cache)
	{
		_options = options?.Value;
		_policies = policies?.Value;
	}

	public async Task SeedAsync()
	{
		// on startup, save the IP rules defined in appsettings
		if (_options != null && _policies != null)
		{
			await SetAsync($"{_options.IpPolicyPrefix}", _policies).ConfigureAwait(false);
		}
	}
}

这个是用例存储IpRateLimitPolicies（ip限制）。

MemoryCacheIpPolicyStore 这个名字起的有点意思，MemoryCache 是内存缓存，IpPolicy ip策略，store 存储。

分别是存储空间、物品、功能的组合。所以这个库应该是外国人写的，一般来说中国人会这样改:IpPolicyMemoryCacheStore,估计是因为强调故而把MemoryCache放到前面去了。

这里我刚开始有点不理解，本来已经可以读取到了options，那么按照options操作就很方便了。

那么为啥要用缓存到内存中呢？后来大体的通读了一下，是因为_policies（特殊制定的ip规则）很多地方都要使用到，一方面是为了解耦，另外一方面呢，是因为下面这个。

[HttpPost]
public void Post()
{
	var pol = _ipPolicyStore.Get(_options.IpPolicyPrefix);

	pol.IpRules.Add(new IpRateLimitPolicy
	{
		Ip = "8.8.4.4",
		Rules = new List<RateLimitRule>(new RateLimitRule[] {
			new RateLimitRule {
				Endpoint = "*:/api/testupdate",
				Limit = 100,
				Period = "1d" }
		})
	});

	_ipPolicyStore.Set(_options.IpPolicyPrefix, pol);
}

是可以动态设置特殊ip的一些配置的。 那么里面也考虑到了分布式的一些行为，比如把缓存放到redis这种隔离缓存中。

如果将_policies 封装到memory cache 中，那么和redis cache形成了一套适配器。个人认为是从设计方面考虑的。

然后看下这个方法，里面就是以IpRateLimiting的IpPolicyPrefix 作为key，然后存储了IpRateLimitPolicies。

public async Task SeedAsync()
{
	// on startup, save the IP rules defined in appsettings
	if (_options != null && _policies != null)
	{
		await SetAsync($"{_options.IpPolicyPrefix}", _policies).ConfigureAwait(false);
	}
}

具体的SetAsync 如下:

public Task SetAsync(string id, T entry, TimeSpan? expirationTime = null, CancellationToken cancellationToken = default)
{
	var options = new MemoryCacheEntryOptions
	{
		Priority = CacheItemPriority.NeverRemove
	};

	if (expirationTime.HasValue)
	{
		options.SetAbsoluteExpiration(expirationTime.Value);
	}

	_cache.Set(id, entry, options);

	return Task.CompletedTask;
}

然后这里值得注意的是_options.IpPolicyPrefix，这个值如果是分布式那么应该值得关注一下，因为我们有不同应用服务，如果希望不同的应用服务用到不同的ip限制，那么IpPolicyPrefix 最好改成应用名，而不是使用默认值。

那么看下MemoryCacheClientPolicyStore：

public class MemoryCacheClientPolicyStore : MemoryCacheRateLimitStore<ClientRateLimitPolicy>, IClientPolicyStore
{
	private readonly ClientRateLimitOptions _options;
	private readonly ClientRateLimitPolicies _policies;

	public MemoryCacheClientPolicyStore(
		IMemoryCache cache,
		IOptions<ClientRateLimitOptions> options = null,
		IOptions<ClientRateLimitPolicies> policies = null) : base(cache)
	{
		_options = options?.Value;
		_policies = policies?.Value;
	}

	public async Task SeedAsync()
	{
		// on startup, save the IP rules defined in appsettings
		if (_options != null && _policies?.ClientRules != null)
		{
			foreach (var rule in _policies.ClientRules)
			{
				await SetAsync($"{_options.ClientPolicyPrefix}_{rule.ClientId}", new ClientRateLimitPolicy { ClientId = rule.ClientId, Rules = rule.Rules }).ConfigureAwait(false);
			}
		}
	}
}

这个就是client id的限制的缓存的，和上面一样就不看了。

MemoryCacheRateLimitCounterStore:

public class MemoryCacheRateLimitCounterStore : MemoryCacheRateLimitStore<RateLimitCounter?>, IRateLimitCounterStore
{
	public MemoryCacheRateLimitCounterStore(IMemoryCache cache) : base(cache)
	{
	}
}

这里面没有啥子。但是从名字上猜测，里面是缓存每个ip请求次数的当然还有时间，主要起缓存作用。

最后一个：AsyncKeyLockProcessingStrategy

public class AsyncKeyLockProcessingStrategy : ProcessingStrategy
{
	private readonly IRateLimitCounterStore _counterStore;
	private readonly IRateLimitConfiguration _config;

	public AsyncKeyLockProcessingStrategy(IRateLimitCounterStore counterStore, IRateLimitConfiguration config)
		: base(config)
	{
		_counterStore = counterStore;
		_config = config;
	}

	/// The key-lock used for limiting requests.
	private static readonly AsyncKeyLock AsyncLock = new AsyncKeyLock();

	public override async Task<RateLimitCounter> ProcessRequestAsync(ClientRequestIdentity requestIdentity, RateLimitRule rule, ICounterKeyBuilder counterKeyBuilder, RateLimitOptions rateLimitOptions, CancellationToken cancellationToken = default)
	{
		var counter = new RateLimitCounter
		{
			Timestamp = DateTime.UtcNow,
			Count = 1
		};

		var counterId = BuildCounterKey(requestIdentity, rule, counterKeyBuilder, rateLimitOptions);

		// serial reads and writes on same key
		using (await AsyncLock.WriterLockAsync(counterId).ConfigureAwait(false))
		{
			var entry = await _counterStore.GetAsync(counterId, cancellationToken);

			if (entry.HasValue)
			{
				// entry has not expired
				if (entry.Value.Timestamp + rule.PeriodTimespan.Value >= DateTime.UtcNow)
				{
					// increment request count
					var totalCount = entry.Value.Count + _config.RateIncrementer?.Invoke() ?? 1;

					// deep copy
					counter = new RateLimitCounter
					{
						Timestamp = entry.Value.Timestamp,
						Count = totalCount
					};
				}
			}

			// stores: id (string) - timestamp (datetime) - total_requests (long)
			await _counterStore.SetAsync(counterId, counter, rule.PeriodTimespan.Value, cancellationToken);
		}

		return counter;
	}
}

估摸着是执行具体计数逻辑的，那么等执行中间件的时候在看。

后面有写入了一个：services.AddSingleton<IRateLimitConfiguration, RateLimitConfiguration>();

那么这个RateLimitConfiguration 是做什么的呢?

public class RateLimitConfiguration : IRateLimitConfiguration
{
	public IList<IClientResolveContributor> ClientResolvers { get; } = new List<IClientResolveContributor>();
	public IList<IIpResolveContributor> IpResolvers { get; } = new List<IIpResolveContributor>();

	public virtual ICounterKeyBuilder EndpointCounterKeyBuilder { get; } = new PathCounterKeyBuilder();

	public virtual Func<double> RateIncrementer { get; } = () => 1;

	public RateLimitConfiguration(
		IOptions<IpRateLimitOptions> ipOptions,
		IOptions<ClientRateLimitOptions> clientOptions)
	{
		IpRateLimitOptions = ipOptions?.Value;
		ClientRateLimitOptions = clientOptions?.Value;
	}

	protected readonly IpRateLimitOptions IpRateLimitOptions;
	protected readonly ClientRateLimitOptions ClientRateLimitOptions;

	public virtual void RegisterResolvers()
	{
		string clientIdHeader = GetClientIdHeader();
		string realIpHeader = GetRealIp();

		if (clientIdHeader != null)
		{
			ClientResolvers.Add(new ClientHeaderResolveContributor(clientIdHeader));
		}

		// the contributors are resolved in the order of their collection index
		if (realIpHeader != null)
		{
			IpResolvers.Add(new IpHeaderResolveContributor(realIpHeader));
		}

		IpResolvers.Add(new IpConnectionResolveContributor());
	}

	protected string GetClientIdHeader()
	{
		return ClientRateLimitOptions?.ClientIdHeader ?? IpRateLimitOptions?.ClientIdHeader;
	}

	protected string GetRealIp()
	{
		return IpRateLimitOptions?.RealIpHeader ?? ClientRateLimitOptions?.RealIpHeader;
	}
}

重点看:

public virtual void RegisterResolvers()
{
	string clientIdHeader = GetClientIdHeader();
	string realIpHeader = GetRealIp();

	if (clientIdHeader != null)
	{
		ClientResolvers.Add(new ClientHeaderResolveContributor(clientIdHeader));
	}

	// the contributors are resolved in the order of their collection index
	if (realIpHeader != null)
	{
		IpResolvers.Add(new IpHeaderResolveContributor(realIpHeader));
	}

	IpResolvers.Add(new IpConnectionResolveContributor());
}

这里只看ip部分:

protected string GetRealIp()
{
	return IpRateLimitOptions?.RealIpHeader ?? ClientRateLimitOptions?.RealIpHeader;
}

那么这个IpHeaderResolveContributor是什么呢？

public class IpHeaderResolveContributor : IIpResolveContributor
{
	private readonly string _headerName;

	public IpHeaderResolveContributor(
		string headerName)
	{
		_headerName = headerName;
	}

	public string ResolveIp(HttpContext httpContext)
	{
		IPAddress clientIp = null;

		if (httpContext.Request.Headers.TryGetValue(_headerName, out var values))
		{
			clientIp = IpAddressUtil.ParseIp(values.Last());
		}

		return clientIp?.ToString();
	}
}

原来是配置是从header的哪个位置获取ip。官网demo中给的是"RealIpHeader": "X-Real-IP"。从header部分的RealIpHeader获取。

同样，官方也默认提供了IpResolvers.Add(new IpConnectionResolveContributor());。

public class IpConnectionResolveContributor : IIpResolveContributor
{

	public IpConnectionResolveContributor()
	{

	}

	public string ResolveIp(HttpContext httpContext)
	{
		return httpContext.Connection.RemoteIpAddress?.ToString();
	}
}

从httpContext.Connection.RemoteIpAddress 中获取ip，那么问题来了，RemoteIpAddress 是如何获取的呢？ 到底X-Real-IP 获取的ip准不准呢？会在.net core 细节篇中介绍。

回到原始。现在已经注入了服务，那么如何把中间件注入进去呢？

在Configure 中:

app.UseIpRateLimiting();

将会执行中间件：IpRateLimitMiddleware

public class IpRateLimitMiddleware : RateLimitMiddleware<IpRateLimitProcessor>
{
	private readonly ILogger<IpRateLimitMiddleware> _logger;

	public IpRateLimitMiddleware(RequestDelegate next,
		IProcessingStrategy processingStrategy,
		IOptions<IpRateLimitOptions> options,
		IRateLimitCounterStore counterStore,
		IIpPolicyStore policyStore,
		IRateLimitConfiguration config,
		ILogger<IpRateLimitMiddleware> logger
	)
		: base(next, options?.Value, new IpRateLimitProcessor(options?.Value, counterStore, policyStore, config, processingStrategy), config)
	{
		_logger = logger;
	}

	protected override void LogBlockedRequest(HttpContext httpContext, ClientRequestIdentity identity, RateLimitCounter counter, RateLimitRule rule)
	{
		_logger.LogInformation($"Request {identity.HttpVerb}:{identity.Path} from IP {identity.ClientIp} has been blocked, quota {rule.Limit}/{rule.Period} exceeded by {counter.Count - rule.Limit}. Blocked by rule {rule.Endpoint}, TraceIdentifier {httpContext.TraceIdentifier}. MonitorMode: {rule.MonitorMode}");
	}
}

查看:RateLimitMiddleware

里面就是具体的invoke中间件代码了。

结

因为篇幅有限，后一节invoke逐行分析其如何实现的。

以上只是个人看源码的过程，希望能得到各位的指点，共同进步。

另外.net core 细节篇整理进度为40%。