zoukankan      html  css  js  c++  java
  • asp.net core mvc剖析:KestrelServer

    KestrelServer是基于Libuv开发的高性能web服务器,那我们现在就来看一下它是如何工作的。在上一篇文章中提到了Program的Main方法,在这个方法里Build了一个WebHost,我们再来看一下代码:

    public static void Main(string[] args)
       {
           var host = new WebHostBuilder()
               .UseKestrel()
               .UseContentRoot(Directory.GetCurrentDirectory())
               .UseIISIntegration()
               .UseStartup<Startup>()
               .Build();
     
           host.Run();
       }
    

      里面有一个UseKestrel方法调用,这个方法的作用就是使用KestrelServer作为web server来提供web服务。在WebHost启动的时候,调用了IServer的Start方法启动服务,由于我们使用KestrelServer作为web server,自然这里调用的就是KestrelServer.Start方法,那我们来看下KestrelServer的Start方法里主要代码:

     首先,我们发现在Start方法里创建了一个KestrelEngine对象,具体代码如下:

    var engine = new KestrelEngine(new ServiceContext
    {
           FrameFactory = context =>
           {
               return new Frame<TContext>(application, context);
           },
           AppLifetime = _applicationLifetime,
           Log = trace,
           ThreadPool = new LoggingThreadPool(trace),
           DateHeaderValueManager = dateHeaderValueManager,
           ServerOptions = Options
     });
    

      KestrelEngine构造方法接受一个ServiceContext对象参数,ServiceContext里包含一个FrameFactory,从名称上很好理解,就是Frame得工厂,Frame是什么?Frame是http请求处理对象,每个请求过来后,都会交给一个Frame对象进行受理,我们这里先记住它的作用,后面还会看到它是怎么实例化的。除了这个外,还有一个是AppLiftTime,它是一个IApplicationLifetime对象,它是整个应用生命周期的管理对象,前面没有说到,这里补充上。

    public interface IApplicationLifetime
        {
            /// <summary>
            /// Triggered when the application host has fully started and is about to wait
            /// for a graceful shutdown.
            /// </summary>
            CancellationToken ApplicationStarted { get; }
    
            /// <summary>
            /// Triggered when the application host is performing a graceful shutdown.
            /// Requests may still be in flight. Shutdown will block until this event completes.
            /// </summary>
            CancellationToken ApplicationStopping { get; }
    
            /// <summary>
            /// Triggered when the application host is performing a graceful shutdown.
            /// All requests should be complete at this point. Shutdown will block
            /// until this event completes.
            /// </summary>
            CancellationToken ApplicationStopped { get; }
    
            /// <summary>
            /// Requests termination the current application.
            /// </summary>
            void StopApplication();
        }
    

      IApplicationLifetime中提供了三个时间点,

      1,ApplicationStarted:应用程序已启动
      2,ApplicationStopping:应用程序正在停止
      3,ApplicationStopped:应用程序已停止

      我们可以通过CancellationToken.Register方法注册回调方法,在上面说到的三个时间点,执行我们特定的业务逻辑。IApplicationLifetime是在WebHost的Start方法里创建的,如果想在我们自己的应用程序获取这个对象,我们可以直接通过依赖注入的方式获取即可。

     我们继续回到ServiceContext对象,这里面还包含了Log对象,用于跟踪日志,一般我们是用来看程序执行的过程,并可以通过它发现程序执行出现问题的地方。还包含一个ServerOptions,它是一个KestrelServerOptions,里面包含跟服务相关的配置参数:

    1,ThreadCount:服务线程数,表示服务启动后,要开启多少个服务线程,因为每个请求都会使用一个线程来进行处理,多线程会提高吞吐量,但是并不一定线程数越多越好,在系统里默认值是跟CPU内核数相等。

    2,ShutdownTimeout:The amount of time after the server begins shutting down before connections will be forcefully closed(在应用程序开始停止到强制关闭当前请求连接所等待的时间,在这个时间段内,应用程序会等待请求处理完,如果还没处理完,将强制关闭)

    3,Limits:KestrelServerLimits对象,里面包含了服务限制参数,比如MaxRequestBufferSize,MaxResponseBufferSize

    其他参数就不再一个一个说明了。

    KestrelEngine对象创建好后,通过调用 engine.Start(threadCount),根据配置的threadcount进行服务线程KestrelThread实例化,代码如下:
         public void Start(int count)
            {
                for (var index = 0; index < count; index++)
                {
                    Threads.Add(new KestrelThread(this));
                }
    
                foreach (var thread in Threads)
                {
                    thread.StartAsync().Wait();
                }
            }

     上面的代码会创建指定数量的Thread对象,然后开始等待任务处理。KestrelThread是对libuv线程处理的封装。

    这些工作都准备好后,就开始启动监听服务了

     foreach (var endPoint in listenOptions)
                    {
                        try
                        {
                            _disposables.Push(engine.CreateServer(endPoint));
                        }
                        catch (AggregateException ex)
                        {
                            if ((ex.InnerException as UvException)?.StatusCode == Constants.EADDRINUSE)
                            {
                                throw new IOException($"Failed to bind to address {endPoint}: address already in use.", ex);
                            }
    
                            throw;
                        }
    
                        // If requested port was "0", replace with assigned dynamic port.
                        _serverAddresses.Addresses.Add(endPoint.ToString());
                    }
    

      上面红色字体代码,就是创建监听服务的方法,我们再详细看下里面的详细情况:

          public IDisposable CreateServer(ListenOptions listenOptions)
            {
                var listeners = new List<IAsyncDisposable>();
    
                try
                {
    //如果前面创建的线程数量为1,直接创建listener对象,启动监听 if (Threads.Count == 1) { var listener = new Listener(ServiceContext); listeners.Add(listener); listener.StartAsync(listenOptions, Threads[0]).Wait(); } else {
                //如果线程数不为1的时候 var pipeName = (Libuv.IsWindows ? @"\.pipekestrel_" : "/tmp/kestrel_") + Guid.NewGuid().ToString("n"); var pipeMessage = Guid.NewGuid().ToByteArray();            //先创建一个主监听对象,这个Listenerprimary就是一个Listener,监听socket就是在这里面创建的 var listenerPrimary = new ListenerPrimary(ServiceContext); listeners.Add(listenerPrimary);
                //启动监听 listenerPrimary.StartAsync(pipeName, pipeMessage, listenOptions, Threads[0]).Wait(); //为剩余的每个服务线程关联一个ListenerSecondary对象,这个对象使用命名Pipe与主监听对象通信,在主监听对象接收到请求后,通过pipe把接受的socket对象发送给特定的线程处理 foreach (var thread in Threads.Skip(1)) { var listenerSecondary = new ListenerSecondary(ServiceContext); listeners.Add(listenerSecondary); listenerSecondary.StartAsync(pipeName, pipeMessage, listenOptions, thread).Wait(); } } return new Disposable(() => { DisposeListeners(listeners); }); } catch { DisposeListeners(listeners); throw; } }

      这个时候服务就开始接受http请求了,我们前面说到了,监听socket在listener类中创建(ListenerPrimary也是一个Listener),下面是listener的start方法

          public Task StartAsync(
                ListenOptions listenOptions,
                KestrelThread thread)
            {
                ListenOptions = listenOptions;
                Thread = thread;
    
                var tcs = new TaskCompletionSource<int>(this);
    
                Thread.Post(state =>
                {
                    var tcs2 = (TaskCompletionSource<int>) state;
                    try
                    {
                        var listener = ((Listener) tcs2.Task.AsyncState);
    //创建监听socket listener.ListenSocket = listener.CreateListenSocket();
    //开始监听,当有连接请求过来后,触发ConnectionCallback方法 ListenSocket.Listen(Constants.ListenBacklog, ConnectionCallback, this); tcs2.SetResult(0); } catch (Exception ex) { tcs2.SetException(ex); } }, tcs); return tcs.Task; }

      ConnectionCallback:当连接请求过来后被触发,在回调方法里,进行连接处理分发,连接分发代码如下:

         protected virtual void DispatchConnection(UvStreamHandle socket)
            {
                var connection = new Connection(this, socket);
                connection.Start();
            }
    

      这个是listener类中的实现,我们前面看到,只有在线程数为1的情况下,才创建Listener对象进行监听,否则创建ListenerPrimary监听,ListenerPrimay里重写了方法,它的实现如下:

         protected override void DispatchConnection(UvStreamHandle socket)
            {
    //这里采用轮询的方式,把连接请求依次分发给不同的线程进行处理 var index = _dispatchIndex++ % (_dispatchPipes.Count + 1); if (index == _dispatchPipes.Count) {
       // base.DispatchConnection(socket); } else { DetachFromIOCP(socket); var dispatchPipe = _dispatchPipes[index];
    //这里就是通过命名pipe,传递socket给特定的线程 var write = new UvWriteReq(Log); write.Init(Thread.Loop); write.Write2( dispatchPipe, _dummyMessage, socket, (write2, status, error, state) => { write2.Dispose(); ((UvStreamHandle)state).Dispose(); }, socket); } }

      好了,连接请求找到处理线程后,后面就可以开始处理工作了。ListenerSecondary里的代码比较复杂,其实最终都会调用下面的代码完成Connection对象的创建

     var connection = new Connection(this, socket);
    connection.Start();
    

      Connection表示的就是当前连接,下面是它的构造方法

    public Connection(ListenerContext context, UvStreamHandle socket) : base(context)
            {
                _socket = socket;
                _connectionAdapters = context.ListenOptions.ConnectionAdapters;
                socket.Connection = this;
                ConnectionControl = this;
    
                ConnectionId = GenerateConnectionId(Interlocked.Increment(ref _lastConnectionId));
    
                if (ServerOptions.Limits.MaxRequestBufferSize.HasValue)
                {
                    _bufferSizeControl = new BufferSizeControl(ServerOptions.Limits.MaxRequestBufferSize.Value, this);
                }
            //创建输入输出socket流
                Input = new SocketInput(Thread.Memory, ThreadPool, _bufferSizeControl);
                Output = new SocketOutput(Thread, _socket, this, ConnectionId, Log, ThreadPool);
    
                var tcpHandle = _socket as UvTcpHandle;
                if (tcpHandle != null)
                {
                    RemoteEndPoint = tcpHandle.GetPeerIPEndPoint();
                    LocalEndPoint = tcpHandle.GetSockIPEndPoint();
                }
            //创建处理frame,这里的framefactory就是前面创建KestrelEngine时创建的工厂
                _frame = FrameFactory(this);
                _lastTimestamp = Thread.Loop.Now();
            }
    

      然后调用Connection的Start方法开始进行处理,这里面直接把处理任务交给Frame处理,Start方法实现:

    public void Start()
            {
                Reset();
           //启动了异步处理任务开始进行处理 _requestProcessingTask = Task.Factory.StartNew( (o) => ((Frame)o).RequestProcessingAsync(),//具体的处理方法 this, default(CancellationToken), TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap(); _frameStartedTcs.SetResult(null); }

      

    RequestProcessingAsync方法里不再详细介绍了,把主要的代码拿出来看一下:
    。。。。。
    //_application就是上一篇文章提到的HostApplication,首先调用CreateContext创建HttpContext对象 var context = _application.CreateContext(this); 。。。。。。
    //进入处理管道 await _application.ProcessRequestAsync(context).ConfigureAwait(false); 。。。。。。

      

    ProcessRequestAsync完成处理后,把结果输出给客户端,好到此介绍完毕。如果有问题,欢迎大家指点。
  • 相关阅读:
    学习日报
    阅读笔记2
    学习日报
    记账本开发7
    记账本开发6
    学习日报
    记账本开发5
    今日总结
    今日总结
    家庭记账本7
  • 原文地址:https://www.cnblogs.com/dxp909/p/6397319.html
Copyright © 2011-2022 走看看