目录
- 目标
- 源码
目标
理解 KestrelServer 如何接收网络请求,网络请求如何转换成 http request context(C# 可识别)
源码
https://github.com/dotnet/aspnetcore/
在目录 aspnetcore\src\Servers\Kestrel\Core\src\Internal 下有一个 KestrelServerImpl
internal class KestrelServerImpl : IServer
在 host 启动的时候调用了 server 的 startup 方法,可以从这个入口开始
public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken) where TContext : notnull
StartAsync 方法主要分为以下三步
async Task OnBind(ListenOptions options, CancellationToken onBindCancellationToken)
{
...
}
AddressBindContext = new AddressBindContext(_serverAddresses, Options, Trace, OnBind);
await BindAsync(cancellationToken).ConfigureAwait(false);
BindAsync 方法中使用 AddressBindContext 进行绑定
await AddressBinder.BindAsync(Options.ListenOptions, AddressBindContext!, cancellationToken).ConfigureAwait(false);
在 AddressBinder 的 BindAsync 方法中创建了多种策略进行绑定
var strategy = CreateStrategy(
listenOptions.ToArray(),
context.Addresses.ToArray(),
context.ServerAddressesFeature.PreferHostingUrls);
...
await strategy.BindAsync(context, cancellationToken).ConfigureAwait(false);
例如 AddressesStrategy,它有自己的一个绑定方法
private class AddressesStrategy : IStrategy
{
protected readonly IReadOnlyCollection<string> _addresses;
public AddressesStrategy(IReadOnlyCollection<string> addresses)
{
_addresses = addresses;
}
public virtual async Task BindAsync(AddressBindContext context, CancellationToken cancellationToken)
{
foreach (var address in _addresses)
{
var options = ParseAddress(address, out var https);
context.ServerOptions.ApplyEndpointDefaults(options);
if (https && !options.IsTls)
{
options.UseHttps();
}
await options.BindAsync(context, cancellationToken).ConfigureAwait(false);
}
}
}
options 来自 IConnectionBuilder 的 ListenOptions 的绑定
public class ListenOptions : IConnectionBuilder, IMultiplexedConnectionBuilder
这一路走下来发现找不到重点,所以需要换一个方向从 OnBind 方法入手,它是一个委托,需要找到调用的地方
async Task OnBind(ListenOptions options, CancellationToken onBindCancellationToken)
可以看到 OnBind 方法传入到 AddressBindContext 中
AddressBindContext = new AddressBindContext(_serverAddresses, Options, Trace, OnBind);
在 AddressBindContext 中它是一个 CreateBinding
public Func<ListenOptions, CancellationToken, Task> CreateBinding { get; }
全局搜索 CreateBinding
可以找到在 AddressBinder 的 BindEndpointAsync 方法中被调用
internal static async Task BindEndpointAsync(ListenOptions endpoint, AddressBindContext context, CancellationToken cancellationToken)
{
try
{
await context.CreateBinding(endpoint, cancellationToken).ConfigureAwait(false);
}
catch (AddressInUseException ex)
{
throw new IOException(CoreStrings.FormatEndpointAlreadyInUse(endpoint), ex);
}
context.ServerOptions.OptionsInUse.Add(endpoint);
}
而 BindEndpointAsync 方法被 ListenOptions 的 BindAsync 方法调用,也就是上面提到的 StartAsync 的第三步 BindAsync 走到的 ListenOptions
internal virtual async Task BindAsync(AddressBindContext context, CancellationToken cancellationToken)
{
await AddressBinder.BindEndpointAsync(this, context, cancellationToken).ConfigureAwait(false);
context.Addresses.Add(GetDisplayName());
}
在第三步 BindAsync 方法中加载配置,加载之后才调用真正的绑定方法
Options.ConfigurationLoader?.Load();
await AddressBinder.BindAsync(Options.ListenOptions, AddressBindContext!, cancellationToken).ConfigureAwait(false);
所以整个过程的重点在第一步的 OnBind,而 OnBind 的重点在于 TransportManager 的 BindAsync 方法
options.EndPoint = await _transportManager.BindAsync(options.EndPoint, multiplexedConnectionDelegate, options, onBindCancellationToken).ConfigureAwait(false);
进入 TransportManager 中可以看到在 BindAsync 方法中开始接收
StartAcceptLoop(new GenericConnectionListener(transport), c => connectionDelegate(c), endpointConfig);
在 StartAcceptLoop 方法中调用了 StartAcceptingConnections 方法
var acceptLoopTask = connectionDispatcher.StartAcceptingConnections(connectionListener);
在 StartAcceptingConnections 方法中将需要被执行的方法添加到队列中
ThreadPool.UnsafeQueueUserWorkItem(StartAcceptingConnectionsCore, listener, preferLocal: false);
在 StartAcceptingConnectionsCore 里面开始监听接收,这就是关键,这里执行了 kestrelConnection,而 kestrelConnection 又包含 _connectionDelegate
var connection = await listener.AcceptAsync();
var kestrelConnection = new KestrelConnection<T>(
id, _serviceContext, _transportConnectionManager, _connectionDelegate, connection, Log);
_transportConnectionManager.AddConnection(id, kestrelConnection);
ThreadPool.UnsafeQueueUserWorkItem(kestrelConnection, preferLocal: false);
在 kestrelConnection 中可以看到整个 ExecuteAsync 方法里面只执行了 _connectionDelegate
await _connectionDelegate(connectionContext);
意识到 _connectionDelegate 的重要性之后再往回找是怎么传进来的,可以找到是在 KestrelServerImpl 中通过 ListenOptions 构建出来的
var connectionDelegate = options.Build();
在 Build 方法里面可以看到它是一个管道
ConnectionDelegate app = context =>
{
return Task.CompletedTask;
};
for (var i = _middleware.Count - 1; i >= 0; i--)
{
var component = _middleware[i];
app = component(app);
}
return app;
通过 _middleware 的 Use 方法的引用找不到有价值的信息
public IConnectionBuilder Use(Func<ConnectionDelegate, ConnectionDelegate> middleware)
{
_middleware.Add(middleware);
return this;
}
于是回到 KestrelServerImpl 中,查看 UseHttpServer 方法
options.UseHttpServer(ServiceContext, application, options.Protocols, addAltSvcHeader);
可以看到这个方法构建了一个 HttpConnectionMiddleware
var middleware = new HttpConnectionMiddleware<TContext>(serviceContext, application, protocols, addAltSvcHeader);
return builder.Use(next =>
{
return middleware.OnConnectionAsync;
});
进入 HttpConnectionMiddleware 可以看到一个核心方法 OnConnectionAsync,创建了一个 HttpConnection,然后调用 ProcessRequestsAsync
var connection = new HttpConnection(httpConnectionContext);
return connection.ProcessRequestsAsync(_application);
在 ProcessRequestsAsync 方法中可以看到 KestrelServer 的核心逻辑,根据不同的协议,执行不同的逻辑;同时可以看到它是如何处理请求的,通过 requestProcessor 处理请求
switch (SelectProtocol())
{
case HttpProtocols.Http1:
requestProcessor = _http1Connection = new Http1Connection<TContext>((HttpConnectionContext)_context);
_protocolSelectionState = ProtocolSelectionState.Selected;
break;
case HttpProtocols.Http2:
requestProcessor = new Http2Connection((HttpConnectionContext)_context);
_protocolSelectionState = ProtocolSelectionState.Selected;
break;
case HttpProtocols.Http3:
requestProcessor = new Http3Connection((HttpMultiplexedConnectionContext)_context);
_protocolSelectionState = ProtocolSelectionState.Selected;
break;
}
await requestProcessor.ProcessRequestsAsync(httpApplication);
requestProcessor 是一个 IRequestProcessor 接口,它有多个实现,以 Http2Connection 为例
internal partial class Http2Connection : IHttp2StreamLifetimeHandler, IHttpStreamHeadersHandler, IRequestProcessor
在 Http2Connection 的 ProcessRequestsAsync 方法中读取流,解析转换,处理
await _frameWriter.WriteWindowUpdateAsync(0, diff);
var result = await Input.ReadAsync();
await ProcessFrameAsync(application, framePayload);
当 HttpConnectionMiddleware 的 OnConnectionAsync 处理完之后,如何与应用层代码拼接,这里只是 Kestrel 的处理
可以通过 IRequestProcessor 接口的 ProcessRequestsAsync 方法的实现找到 HttpProtocol 的 ProcessRequestsAsync 方法,可以看到它执行了一个 ProcessRequests 方法
await ProcessRequests(application);
在 ProcessRequests 方法中将从 Body 里面获取的内容封装到一个 context,这个才是真正的 HttpContext,然后再运行应用层代码,之前都是 Kestrel 的解析逻辑,这里才是串联到我们构建的管道 application
InitializeBodyControl(messageBody);
var context = application.CreateContext(this);
// Run the application code for this request
await application.ProcessRequestAsync(context);
接下来看一下 application 是如何传过来的,一直找到 HttpConnection,HttpConnectionMiddleware,HttpConnectionBuilderExtensions,KestrelServerImpl
public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken) where TContext : notnull
可以看到是 Host 调用 Server 的 StartAsync 传进来的,这里体现了职责分离的原则,对于应用层的管道,定义了一个 IHttpApplication
从 Host 传到 Server,Server 完成了网络端口的绑定,网络的监听接收,网络二进制转换成具体的 c# 可识别的 HTTPContext 之后,调用了 Host 那边封装好的一个 application 应用层的管道,这是 Host 在 Startup 里面定义的,这就是一个完整的过程
课程链接
https://appsqsyiqlk5791.h5.xiaoeknow.com/v1/course/video/v_5f39bdb8e4b01187873136cf?type=2
本作品采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可。
欢迎转载、使用、重新发布,但务必保留文章署名 郑子铭 (包含链接: http://www.cnblogs.com/MingsonZheng/ ),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。
如有任何疑问,请与我联系 (MingsonZheng@outlook.com) 。