kestrel Server的源码分析
NeilHu 人气:0今天这一篇博客讲的是.net core 自带的kestrel server，当你开发微服务k8s部署在linux环境下，一般默认开启这个高性能服务，如果大家之前看过我的owin katana的博客，会发现.net core 的好多实现在之前.net standard 的版本已经实现过了，当时开发的asp.net 程序与IIS紧紧耦合在一起，后来的微软团队意识到这个问题并尝试将asp.net 解耦server，制定了owin标准并启动了一个开源项目katana，这个项目的结果并没有带动社区效应，但是此时微软已经制订了.net core的开发，并在katana文档暗示了.net vnext 版本，这个就是。net core 与owin katana 的故事。强烈建议大家有时间看看owin katana,里面有一些 dependency inject, hash map, raw http 协议等等实现。非常收益。说到这些我们开始步入正题吧。原代码在github上的asp.net core 源码。
上图大致地描述了一个asp.net core 的请求过程，但是我们会发现appication 依赖了server，所以我们需要一个Host 的去解耦server 和aplication 的实现，只要server符合host标准可以任意更换，解耦之后的代码与下图所示。
所以我们的代码都是创建一个web host然后使用usekestrel,如下所示。
var host = new WebHostBuilder()
.UseKestrel(options =>
{
options.Listen(IPAddress.Loopback, 5001);
})
.UseStartup<Startup>();
首先我们知道一个server 实现需要网络编程，所以我们需要socket库来快速编程，它已经帮你实现了tcp与udp协议，不需要自己重复的造轮子。首先我们需要看UseKestrel的方法做了什么。
1 public static IWebHostBuilder UseKestrel(this IWebHostBuilder hostBuilder)
2 {
3 return hostBuilder.ConfigureServices(services =>
4 {
5 // Don't override an already-configured transport
6 services.TryAddSingleton<IConnectionListenerFactory, SocketTransportFactory>();
7
8 services.AddTransient<IConfigureOptions<KestrelServerOptions>, KestrelServerOptionsSetup>();
9 services.AddSingleton<IServer, KestrelServer>();
10 });
11 }
依赖注入注册了三个对象，一个连接池，一个配置类还有一个是server，会和web host注册了IServer 的实现类,然后我们继续看一下，当你调用run的时候会将控制权从web host 转移给server，如下代码第18行所示。
1 public virtual async Task StartAsync(CancellationToken cancellationToken = default)
2 {
3 HostingEventSource.Log.HostStart(); 6
7 var application = BuildApplication();
8
12 // Fire IHostedService.Start
13 await _hostedServiceExecutor.StartAsync(cancellationToken).ConfigureAwait(false);//启动后台服务
14
15 var diagnosticSource = _applicationServices.GetRequiredService<DiagnosticListener>();
16 var httpContextFactory = _applicationServices.GetRequiredService<IHttpContextFactory>();
17 var hostingApp = new HostingApplication(application, _logger, diagnosticSource, httpContextFactory);
18 await Server.StartAsync(hostingApp, cancellationToken).ConfigureAwait(false);//socket 启动
19 _startedServer = true;
20
21 // Fire IApplicationLifetime.Started
22 _applicationLifetime?.NotifyStarted();
23
24
25 _logger.Started();
26
27 // Log the fact that we did load hosting startup assemblies.
28 if (_logger.IsEnabled(LogLevel.Debug))
29 {
30 foreach (var assembly in _options.GetFinalHostingStartupAssemblies())
31 {
32 _logger.LogDebug("Loaded hosting startup assembly {assemblyName}", assembly);
33 }
34 }
35
36 if (_hostingStartupErrors != null)
37 {
38 foreach (var exception in _hostingStartupErrors.InnerExceptions)
39 {
40 _logger.HostingStartupAssemblyError(exception);
41 }
42 }
43 }
当我们转进到StartAsync方法时会看到如下代码
1 public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken)
2 {
3 try
4 {
19 // ServiceContext.Heartbeat?.Start();//一个是连接池一个是日期时间
20
21 async Task OnBind(ListenOptions options)
22 {
23 // Add the HTTP middleware as the terminal connection middleware
24 options.UseHttpServer(ServiceContext, application, options.Protocols);//注册中间件
25
26 var connectionDelegate = options.Build();
27
28 // Add the connection limit middleware
29 if (Options.Limits.MaxConcurrentConnections.HasValue)
30 {
31 connectionDelegate = new ConnectionLimitMiddleware(connectionDelegate, Options.Limits.MaxConcurrentConnections.Value, Trace).OnConnectionAsync;
32 }
33
34 var connectionDispatcher = new ConnectionDispatcher(ServiceContext, connectionDelegate);
35 var transport = await _transportFactory.BindAsync(options.EndPoint).ConfigureAwait(false);
36
37 // Update the endpoint
38 options.EndPoint = transport.EndPoint;
39 var acceptLoopTask = connectionDispatcher.StartAcceptingConnections(transport);
40
41 _transports.Add((transport, acceptLoopTask));
42 }
43
44 await AddressBinder.BindAsync(_serverAddresses, Options, Trace, OnBind).ConfigureAwait(false);
45 }
46 catch (Exception ex)
47 {51 }
52 }
AddressBinder就是server绑定的ip地址，这个可以在StartUp方法或者环境变量里面配置，里面传了一个回调方法OnBind, 在第24行的UseHttpServer会注册server 内部的中间件去处理这个请求，在第35行socet会绑定地址，用tcp协议，默认使用512个最大pending队列，在接受socket会有多处异步编程和开启线程，建议大家在调试的时候可以修改代码用尽可能少的线程来进行调试。accept 的代码如下图所示
1 private void StartAcceptingConnectionsCore(IConnectionListener listener)
2 {
3 // REVIEW: Multiple accept loops in parallel?
4 _ = AcceptConnectionsAsync();
5
6 async Task AcceptConnectionsAsync()
7 {
8 try
9 {
10 while (true)
11 {
12 var connection = await listener.AcceptAsync();
13 19
20 // Add the connection to the connection manager before we queue it for execution
21 var id = Interlocked.Increment(ref _lastConnectionId);
22 var kestrelConnection = new KestrelConnection(id, _serviceContext, _connectionDelegate, connection, Log);
23
24 _serviceContext.ConnectionManager.AddConnection(id, kestrelConnection);27
28 ThreadPool.UnsafeQueueUserWorkItem(kestrelConnection, preferLocal: false);
29 }
30 }
31 catch (Exception ex)
32 {
33 // REVIEW: If the accept loop ends should this trigger a server shutdown? It will manifest as a hang
34 Log.LogCritical(0, ex, "The connection listener failed to accept any new connections.");
35 }
36 finally
37 {
38 _acceptLoopTcs.TrySetResult(null);
39 }
40 }
41 }
接收到accept socket的时候，会创建一个kestrelconnection 对象，这个对象实现线程方法，然后它会重新去等待一个请求的到来，而用户代码的执行则交给线程池执行。在第14行就是之前kerstrel server 内部的中间件build生成的方法，他的主要功能就是解析socket的携带http信息。
1 internal async Task ExecuteAsync()
2 {
3 var connectionContext = TransportConnection;
4
5 try
6 {
10 using (BeginConnectionScope(connectionContext))
11 {
12 try
13 {
14 await _connectionDelegate(connectionContext);
15 }
16 catch (Exception ex)
17 {
18 Logger.LogError(0, ex, "Unhandled exception while processing {ConnectionId}.", connectionContext.ConnectionId);
19 }
20 }
21 }
22 finally
23 {
34 _serviceContext.ConnectionManager.RemoveConnection(_id);
35 }
36 }
由于http协议版本的不一致导致解析方式的不同，如果有兴趣的小伙伴可以具体查看这一块的逻辑。
1 switch (SelectProtocol())
2 {
3 case HttpProtocols.Http1:
4 // _http1Connection must be initialized before adding the connection to the connection manager
5 requestProcessor = _http1Connection = new Http1Connection<TContext>(_context);
6 _protocolSelectionState = ProtocolSelectionState.Selected;
7 break;
8 case HttpProtocols.Http2:
9 // _http2Connection must be initialized before yielding control to the transport thread,
10 // to prevent a race condition where _http2Connection.Abort() is called just as
11 // _http2Connection is about to be initialized.
12 requestProcessor = new Http2Connection(_context);
13 _protocolSelectionState = ProtocolSelectionState.Selected;
14 break;
15 case HttpProtocols.None:
16 // An error was already logged in SelectProtocol(), but we should close the connection.
17 break;
18 default:
19 // SelectProtocol() only returns Http1, Http2 or None.
20 throw new NotSupportedException($"{nameof(SelectProtocol)} returned something other than Http1, Http2 or None.");
21 }
然后server解析完请求之后所做的重要的一步就是创建httpContext,然后server在第40行将控制权转给web host，web host 会自动调用application code 也就是用户代码。
1 private async Task ProcessRequests<TContext>(IHttpApplication<TContext> application)
2 {
3 while (_keepAlive)
4 {
33 var context = application.CreateContext(this);
34
35 try
36 {
37 KestrelEventSource.Log.RequestStart(this);
38
39 // Run the application code for this request
40 await application.ProcessRequestAsync(context);
41 55 }
56 catch (BadHttpRequestException ex)
57 {
58 // Capture BadHttpRequestException for further processing
59 // This has to be caught here so StatusCode is set properly before disposing the HttpContext
60 // (DisposeContext logs StatusCode).
61 SetBadRequestState(ex);
62 ReportApplicationError(ex);
63 }
64 catch (Exception ex)
65 {
66 ReportApplicationError(ex);
67 }
68
69 KestrelEventSource.Log.RequestStop(this);129 }
130 }
到这里server 的工作大部分都结束了，在之前的描述中我们看到web host 怎么将控制权给到server 的， server 创建好httpContext规则后又是如何将控制权给到web host ， web host 又如何去调用application code的， web host 实际上build 的时候将用户的中间件定义为链表结构暴露一个入口供web host调用，其他的有时间我会再写博客描述这一块。谢谢大家今天的阅读了。欢迎大家能够留言一起讨论。最后谢谢大家的阅读，如果有任何不懂的地方可以私信我。
加载全部内容