flask源码(1)__梳理下简单的流程。浅析app.run的底层原理

我就是我 2022-05-28 05:27 84阅读 0赞

## 前景 ##

这篇文章的主旨在于对本地开发服务器的启动流程的理解，争取能够一窥服务器底层的启动。

## 一个app.run都干了些什么？ ##

### run方法 ###

首先就是Flask.app.run方法

def run(self, host=None, port=None, debug=None, **options):
        """ 不要在生产环境中用run()，因为它不是用来给生产服务器提供安全和执行需求的。 debug标记被设置，那么服务器就会自动重载代码的变动，一旦出现异常就会，显示一个debugger，如果想在debug模式运行，但是又想关闭交互式的debugger，设置use_evalex=False; 并不建议把这个函数用在自动重载的开发上，因为它支持不好。 警告： Flask 会抑制任何服务器错误页面的生成，除非是debug模式，比如，你想用交互式debug，而不让代码重载，你可以设置debug=Tru，use_reloader=Flase. """
        from werkzeug.serving import run_simple
            if host is None:
                host = '127.0.0.1'
            if port is None:
                server_name = self.config['SERVER_NAME']
                if server_name and ':' in server_name:
                    port = int(server_name.rsplit(':', 1)[1])
                else:
                    port = 5000
            if debug is not None:
                self.debug = bool(debug)
            options.setdefault('use_reloader', self.debug)
            options.setdefault('use_debugger', self.debug)
            try:
                run_simple(host, port, self, **options)
            finally:
                self._got_first_request = False

**解读一下：**  
我们可以看到当运行app的时候，是从werkzeug中导入了一个方法run\_simple。前面的是默认host和默认port，最后try 调用run\_simple，启动一个WSGI的应用

### run\_simple ###

def run_simple(hostname, port, application, use_reloader=False, use_debugger=False, use_evalex=True, extra_files=None, reloader_interval=1, reloader_type='auto', threaded=False, processes=1, request_handler=None, static_files=None, passthrough_errors=False, ssl_context=None):
        # 检查端口号是否是整数
        if not isinstance(port, int):
            raise TypeError('port must be an integer')
        # 检查是否用了debugger
        if use_debugger:
            from werkzeug.debug import DebuggedApplication
            application = DebuggedApplication(application, use_evalex)
        # 检查是否用了static_files
        if static_files:
            from werkzeug.wsgi import SharedDataMiddleware
            application = SharedDataMiddleware(application, static_files)
    
        def log_startup(sock):
    
    
        def inner():
            try:
                fd = int(os.environ['WERKZEUG_SERVER_FD'])
            except (LookupError, ValueError):
                fd = None
            srv = make_server(hostname, port, application, threaded,
                              processes, request_handler,
                              passthrough_errors, ssl_context,
                              fd=fd)
            if fd is None:
                log_startup(srv.socket)
            srv.serve_forever()
    
        if use_reloader：
    
        else:
            inner()

**解读一下:**  
OK，由于我是新手，我就按照最简单的来吧，就是默认的那种，可以看到，调用的a是inner()，看到了熟悉的身影:make\_server(),srv.serve\_forever()昨天刚见的，简易的WSGI，让我们再进去看看

#### make\_server ####

def make_server(host=None, port=None, app=None, threaded=False, processes=1, request_handler=None, passthrough_errors=False, ssl_context=None, fd=None):
        """Create a new server instance that is either threaded, or forks or just processes one request after another. """
        if threaded and processes > 1:
            raise ValueError("cannot have a multithreaded and "
                             "multi process server.")
        elif threaded:
            return ThreadedWSGIServer(host, port, app, request_handler,
                                      passthrough_errors, ssl_context, fd=fd)
        elif processes > 1:
            return ForkingWSGIServer(host, port, app, processes, request_handler,
                                     passthrough_errors, ssl_context, fd=fd)
        else:
            return BaseWSGIServer(host, port, app, request_handler,
                                  passthrough_errors, ssl_context, fd=fd)

**解读一下**  
妈耶0.0，再次取最简单的，BaseWSGIServer。

### 服务器部分，Server ###

服务器部分的继承顺序大致为`BaseWSGIServer->HTTPServer->socketserver.TCPServer->socketserver.BaseServer`,我们从最上面一层开始看

#### socketserver.BaseServer ####

class BaseServer:
    
        """服务器类型的基类型. 被调用的方法: - __init__(server_address, RequestHandlerClass) - serve_forever(poll_interval=0.5) - shutdown() - handle_request() # if you do not use serve_forever() - fileno() -> int # for selector 方法重写部分: - server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - server_close() - process_request(request, client_address) - shutdown_request(request) - close_request(request) - service_actions() - handle_error() 子类的方法: - finish_request(request, client_address) 可能被衍生出来的类或者实例重写的变量: - timeout - address_family - socket_type - allow_reuse_address 实例变量: - RequestHandlerClass - socket """
    
        timeout = None
    
        def __init__(self, server_address, RequestHandlerClass):
            """构造器，可能被扩展但是不会被重写"""
            self.server_address = server_address
            self.RequestHandlerClass = RequestHandlerClass
            self.__is_shut_down = threading.Event()
            self.__shutdown_request = False
    
        def server_activate(self):
            """被构造器调用，然后激活服务器. 可能被重写. """
            pass
    
        def serve_forever(self, poll_interval=0.5):
            """处理服务器，直到被关闭. 每隔0.5s轮询一次 """
            self.__is_shut_down.clear()
            try:
                # XXX: Consider using another file descriptor or connecting to the
                # socket to wake this up instead of polling. Polling reduces our
                # responsiveness to a shutdown request and wastes cpu at all other
                # times.
                with _ServerSelector() as selector:
                    selector.register(self, selectors.EVENT_READ)
    
                    while not self.__shutdown_request:
                        ready = selector.select(poll_interval)
                        if ready:
                            self._handle_request_noblock()
    
                        self.service_actions()
            finally:
                self.__shutdown_request = False
                self.__is_shut_down.set()
    
        def shutdown(self):
            """Stops the serve_forever loop. Blocks until the loop has finished. This must be called while serve_forever() is running in another thread, or it will deadlock. """
            self.__shutdown_request = True
            self.__is_shut_down.wait()
    
        def service_actions(self):
            """Called by the serve_forever() loop. May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop. """
            pass
    
        def handle_request(self):
            """可能以阻塞的方式处理一个请求. 遵循time_out. """
            timeout = self.socket.gettimeout()
            if timeout is None:
                timeout = self.timeout
            elif self.timeout is not None:
                timeout = min(timeout, self.timeout)
            if timeout is not None:
                deadline = time() + timeout
    
            # 等待请求直到超时
            with _ServerSelector() as selector:
                selector.register(self, selectors.EVENT_READ)
    
                while True:
                    ready = selector.select(timeout)
                    if ready:
                        return self._handle_request_noblock()
                    else:
                        if timeout is not None:
                            timeout = deadline - time()
                            if timeout < 0:
                                return self.handle_timeout()
    
        def _handle_request_noblock(self):
            """ 非阻塞处理一个请求，在这个方法调用之前，假定select已经返回一个可以读的套接字,所以get_request就不会阻塞了，get_request函数将由子类实现 """
            try:
                request, client_address = self.get_request()
            except OSError:
                return
            if self.verify_request(request, client_address):
                try:
                    self.process_request(request, client_address)
                except:
                    self.handle_error(request, client_address)
                    self.shutdown_request(request)
            else:
                self.shutdown_request(request)
    
        def handle_timeout(self):
            """Called if no new request arrives within self.timeout. Overridden by ForkingMixIn. """
            pass
    
        def verify_request(self, request, client_address):
            """Verify the request. May be overridden. Return True if we should proceed with this request. """
            return True
    
        def process_request(self, request, client_address):
            """Call finish_request. Overridden by ForkingMixIn and ThreadingMixIn. """
            self.finish_request(request, client_address)
            self.shutdown_request(request)
    
        def server_close(self):
            """Called to clean-up the server. May be overridden. """
            pass
    
        def finish_request(self, request, client_address):
            """Finish one request by instantiating RequestHandlerClass."""
            self.RequestHandlerClass(request, client_address, self)
    
        def shutdown_request(self, request):
            """Called to shutdown and close an individual request."""
            self.close_request(request)
    
        def close_request(self, request):
            """Called to clean up an individual request."""
            pass
    
        def handle_error(self, request, client_address):
            """Handle an error gracefully. May be overridden. The default is to print a traceback and continue. """
            print('-'*40)
            print('Exception happened during processing of request from', end=' ')
            print(client_address)
            import traceback
            traceback.print_exc() # XXX But this goes to stderr!
            print('-'*40)

emmm，由于我本身的能力不够，相信关于socket和http这些底层实现对我来说，理解是比较难的，所以我先忽略这部分，直接去看WSGI把，BaseWSGIServer里面处理请求的关键语句就是`handler=WSGIRequestHandler`

### Request处理–WSGIRequestHandler ###

class WSGIRequestHandler(BaseHTTPRequestHandler, object):
    
        """一个request的处理，实现WSGI的分发"""
        def make_environ(self):
            request_url = url_parse(self.path)
    
            def shutdown_server():
                self.server.shutdown_signal = True
    
            url_scheme = self.server.ssl_context is None and 'http' or 'https'
            path_info = url_unquote(request_url.path)
    
            environ = {
                'wsgi.version':         (1, 0),
                'wsgi.url_scheme':      url_scheme,
                'wsgi.input':           self.rfile,
                'wsgi.errors':          sys.stderr,
                'wsgi.multithread':     self.server.multithread,
                'wsgi.multiprocess':    self.server.multiprocess,
                'wsgi.run_once':        False,
                'werkzeug.server.shutdown': shutdown_server,
                'SERVER_SOFTWARE':      self.server_version,
                'REQUEST_METHOD':       self.command,
                'SCRIPT_NAME':          '',
                'PATH_INFO':            wsgi_encoding_dance(path_info),
                'QUERY_STRING':         wsgi_encoding_dance(request_url.query),
                'REMOTE_ADDR':          self.address_string(),
                'REMOTE_PORT':          self.port_integer(),
                'SERVER_NAME':          self.server.server_address[0],
                'SERVER_PORT':          str(self.server.server_address[1]),
                'SERVER_PROTOCOL':      self.request_version
            }
    
            for key, value in self.headers.items():
                key = key.upper().replace('-', '_')
                if key not in ('CONTENT_TYPE', 'CONTENT_LENGTH'):
                    key = 'HTTP_' + key
                environ[key] = value
    
            if environ.get('HTTP_TRANSFER_ENCODING', '').strip().lower() == 'chunked':
                environ['wsgi.input_terminated'] = True
                environ['wsgi.input'] = DechunkedInput(environ['wsgi.input'])
    
            if request_url.scheme and request_url.netloc:
                environ['HTTP_HOST'] = request_url.netloc
    
            return environ
    
        def run_wsgi(self):
            if self.headers.get('Expect', '').lower().strip() == '100-continue':
                self.wfile.write(b'HTTP/1.1 100 Continue\r\n\r\n')
            # WSGI服务器创建environ
            self.environ = environ = self.make_environ()
            headers_set = []
            headers_sent = []
    
            def write(data):
                assert headers_set, 'write() before start_response'
                if not headers_sent:
                    status, response_headers = headers_sent[:] = headers_set
                    try:
                        code, msg = status.split(None, 1)
                    except ValueError:
                        code, msg = status, ""
                    code = int(code)
                    self.send_response(code, msg)
                    header_keys = set()
                    for key, value in response_headers:
                        self.send_header(key, value)
                        key = key.lower()
                        header_keys.add(key)
                    if not ('content-length' in header_keys or
                            environ['REQUEST_METHOD'] == 'HEAD' or
                            code < 200 or code in (204, 304)):
                        self.close_connection = True
                        self.send_header('Connection', 'close')
                    if 'server' not in header_keys:
                        self.send_header('Server', self.version_string())
                    if 'date' not in header_keys:
                        self.send_header('Date', self.date_time_string())
                    self.end_headers()
    
                assert isinstance(data, bytes), 'applications must write bytes'
                self.wfile.write(data)
                self.wfile.flush()
    
            def start_response(status, response_headers, exc_info=None):
                if exc_info:
                    try:
                        if headers_sent:
                            reraise(*exc_info)
                    finally:
                        exc_info = None
                elif headers_set:
                    raise AssertionError('Headers already set')
                headers_set[:] = [status, response_headers]
                return write
    
            def execute(app):
                application_iter = app(environ, start_response)
                try:
                    for data in application_iter:
                        write(data)
                    if not headers_sent:
                        write(b'')
                finally:
                    if hasattr(application_iter, 'close'):
                        application_iter.close()
                    application_iter = None
    
            try:
                execute(self.server.app)
            except (socket.error, socket.timeout) as e:
                self.connection_dropped(e, environ)
            except Exception:
                if self.server.passthrough_errors:
                    raise
                from werkzeug.debug.tbtools import get_current_traceback
                traceback = get_current_traceback(ignore_system_exceptions=True)
                try:
                    # if we haven't yet sent the headers but they are set
                    # we roll back to be able to set them again.
                    if not headers_sent:
                        del headers_set[:]
                    execute(InternalServerError())
                except Exception:
                    pass
                self.server.log('error', 'Error on request:\n%s',
                                traceback.plaintext)
    
        def handle(self):
            """Handles a request ignoring dropped connections."""
            rv = None
            try:
                rv = BaseHTTPRequestHandler.handle(self)
            except (socket.error, socket.timeout) as e:
                self.connection_dropped(e)
            except Exception:
                if self.server.ssl_context is None or not is_ssl_error():
                    raise
            if self.server.shutdown_signal:
                self.initiate_shutdown()
            return rv
    
        def handle_one_request(self):
            """Handle a single HTTP request."""
            self.raw_requestline = self.rfile.readline()
            if not self.raw_requestline:
                self.close_connection = 1
            # 如果解析成功
            elif self.parse_request():
                return self.run_wsgi()
    
        def send_response(self, code, message=None):
            """Send the response header and log the response code."""
            self.log_request(code)
            if message is None:
                message = code in self.responses and self.responses[code][0] or ''
            if self.request_version != 'HTTP/0.9':
                hdr = "%s %d %s\r\n" % (self.protocol_version, code, message)
                self.wfile.write(hdr.encode('ascii'))

**解读一下：**  
run\_wsgi,主要提供的就是execute，也就是app调用了，WSGI将自己上面写好的环境和start\_response函数，当做参数调用app这个函数，而app就是我们编写的那个app

## 总结： ##

**第一步**  
作为初学者我忽略socket，tcp，http协议的底层实现- -，说实话，我真的看不懂，就理论而言，我都还不清楚，就去看最底层实现。有些难度，所以这部分我直接用一句话总结，接收到request之后，http服务器解析一下，然后将结果送给WSGI服务器，这样理解应该是正确的

**第二步**  
之后的关键点就在于WSGI服务器对request的处理了，也就是WSGIRequestHandler，这部分。这个类将环境和start\_response作为参数传递给app，这个方法。app其实是一个类，由于实现了\_\_call\_\_，所以能被调用。它主要就是生成environ，和start\_response。

附上自己理解的流程图吧  
![这里写图片描述][70]

[70]: /images/20220528/3df3237200904261b35d4f145fe46e5f.png