Tornado 异步框架学习

Tornado是一个很是优秀的非阻塞的web服务框架。按照功能能够把module划分为：

Core web framework：

    web

        Application （Web Application 负责path解析与请求分发，重要的方法'__call__'）

        RequestHandler （请求处理基本，重载http verb 处理请求）

    httpserver  （初始化TCPServer 与 路由读写请求）

Asynchronous networking：

    ioloop （调度callback 以及轮询 socket fd，处理相应的读写或者错误事件）

    iostream （异步的读写处理）

    netutil （建立绑定套接字，监听端口，等待连接 ）

Other utilities

Tornado的异步实现就是创建在ioloop（轮询socket fd 读写事件）和iostream（异步的读写）的基础之上。

因为不一样的os使用的轮询模型不一样，只选出linux内核支持的epoll模型来描述。

    We use epoll (Linux) or kqueue (BSD and Mac OS X; requires python 2.6+) if they are available, 

    or else we fall back on select(). If you are implementing a system that needs to handle thousands of

    simultaneous connections, you should use a system that supports either epoll or queue.

最好的学习参考资料就是代码，下面根据框架正常的使用流程，提供一个各个模块和方法的信息图：

    

import tornado.web
import tornado.ioloop

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        return 'Hello World'

application = tornado.web.Application([(r'/', MainHandler),],)

application.listen(8080)
tornado.ioloop.IOLoop.instance().start()

Application                                             HTTPServer
    __init__    
    add_handler(pattern, handler)
    listen(port, address, **kwargs)
        server = HTTPServer(self, **kwargs)             __init__(request_callback, ...)
        |                                                   self.request_callback = request_callback
        v                                                   TCPServer.__init__()
        server.listen(port, address)                    handle_stream(stream, address, self.request_callback)
    __call__(self, request)                                 HTTPConnection(...)

TCPServer                                                               HTTPConnection
    __init__()                                                              __init__
    listen(port, address)                                                       self.request_callback = request_callback
        sockets = bind_socket(port, address)                                    self._header_callback = stack_context.wrap(self._on_headers)
            #getaddrinfo,create socket, setsockopt                              self.stream.read_until(r'\r\n\r\n', self._header_callback)
            #setblocking(0), bind, listen                                       
        add_sockets(sockets)                                                _on_headers(self, data)
            self.io_loop = IOLoop.instance()                                    # parse http protocol
            add_accept_handler(sock, self._handle_connection,                   eol = data.find(r'\r\n')
                    self.io_loop)                                               start_line = data[:eol]
                def accept_handler(fd, events)                                  method, uri, version = start_line.split(' ')
                    while True:                                                 headers = httputil.HTTPHeaders.parse(data[eol:])
                        connection, address = sock.accept()                     # construct a request object 
                                                                                self._request = HTTPRequest(connection=self, method, uri, version, headers, remote_address)
                        self._handle_connection(connection, address)             content-length = headers.get('Content-Length')
                            #If ssl_options wrap socket                         if content-length:
                            stream = IOStream(connection, self.io_loop)             self.stream.read_bytes(content-length, self._on_request_body)
                            #HTTPServer implement handle_stream                     return
                            self.handle_stream(stream, address)                 self.request_callback(self._request)
                                -> HTTPServer.handle_stream                 _on_request_body(self, data)
                self.io_loop.add_handler(sock.fileno(), accept_handler,         # parse content body, and set thire info to self._request.arguments
                                        IOLoop.READ)                            self.request_callback(self._request)

TcpServer调用一个很是重要的方法add_accept_handler，把链接回调函数和相应的socket传递进去，最终

调用ioloop对象注册socket fd 以及对应的callback操做和事件。同时也是在这里把no-blocking socket与异步的

读写经过IOStream链接起来。

IOLoop                                                                      IOStream
    __init__(self, impl=None) #Singleton                                        __init__(conn, self.io_loop)  
                                                                                    #max_buffer_size = 100MB, read_chunk_size=4096 byte
        self._impl = impl or _poll()                                                self._read_buffer = collections.deque()
                                                                                    self._write_buffer = collections.deque()
        self._waker = Waker()                                                   read_until(delimiter, callback)
        self.add_handler(self._waker.fileno(),                                      self._read_delimiter = delimiter
                lambda fd, events: self._waker.consume(),                           self._read_callback = callback
                self.READ)                                                          while True:
                                                                                        if self._read_from_buffer():
    add_handler(fd, handler, events)                                                        return 
        self._handlers[fd] = stack_context.wrap(handler)                                self.check_closed()
        self._impl.register(fd, events|IOLoop.ERROR)                                    if self._read_to_buffer() == 0:
                                                                                            break
    add_callback(callback)                                                              self._add_io_state(self.io_loop.READ)
        self._callbacks.append(stack_context(callback))                                                                        
                                                                                _read_to_buffer()
    start()                                                                         chunk = self.socket.recv(read_chunk_size) 
        while True:                                                                  self._read_buffer.append(chunk)
            poll_timeout = 0.2                                                       self._read_buffer_size += len(chunk)
            with self._callbacks_lock:
                callbacks, self._callbacks = self._callbacks, []                     # Check read buffer size
            for callback in callbacks:                                               return len(chunk)  
                self._run_callback(callback)                                    _read_from_buffer() 
                                                                                    _merge_prefix(self._read_buffer, sys.maxint)
            #deal with timeout handler                                              loc = self._read_buffer[0].find(self._read_delimiter)
            event_pairs = self._imol.poll(poll_timeout)                             if loc != -1:
            self._events.update(event_pairs)                                            callback = self._read_callback
            while self._events:                                                         # reset read control variable
                fd, event = self._events.popitem()                                      self._run_callback(callback, self._consume(loc + delimiter_len))
                self._handlers[fd](fd, event)                                           return True
                                                                                _run_callback(self, callback, *args)
                                                                                    def wrapper():
                                                                                        callback(*args)
                                                                                    with stack_context.NullContext():
                                                                                        self.io_loop.add_callback(wrapper)

在调用IOLoop的 start方法后，IOLoop对象开始循环处理 延迟的数据处理callback 以及timeout callback。最终轮询socket fd

处理fd对应的callback方法。start方法处理的self._callback，到底是什么，在什么时间添加的呢？ 等下咱们会慢慢的揭开面纱。

假设在self._imol.poll(poll_timeout)时，检测到有一个fd有read事件，即有客户链接，则调用对应的callback方法（TCPServer对象的accept_handler对象。），Accept 链接而且紧跟着建立IOStream对象（根据ssl_options选项肯定是不是加密socket）。而后调用handle_stream（因为handle_stream 是在HTTPServer定义，实际上调用的是这里的实现）. 建立一个HTTPConnection对象。在HTTPConnection的初始化函数__init__()中，调用IOStream read_until方法开始读取数据。首先判断读取的数据是否

已经知足条件,若是不知足则继续从socket缓存读取数据.若是已经知足条件,则调用_run_callback,把相应的处理函数(self._header_callback)append到IOLoop对象中,延迟到下次轮询再进行处理. 官方也给出了这样操做的缘由:

        # * Prevents unbounded stack growth when a callback calls an IOLoop operation that immediately runs another callback

        # * Provides a predictable execution context for e.g.  non-reentrant mutexes

        # * Ensures that the try/except in wrapper() is run outside of the application's StackContexts

当下次轮询处理到这个callback时, 执行以前传递_header_callback函数对象, 建立HTTPRequest对象,

若是请求信息包含body则处理body数据块, 过程相似header处理. 当解析完header和body(body可选)以后,

 HTTPConnection对象调用request_callback方法,传递HTTPRequest对象. 这里的request_callback对象即咱们

以前建立的Application对象,这也是为何Application要定义__call__方法.__call__方法根据解析到得path和method

路由到相应的RequestHandler对象,动态的获取相应的方法对象,调用相应的方法. 

Response 过程和read差异不大,实际上HTTPRequest对象包含着HTTPConnection对象,最终的写入操做被传递给IOStream的

write方法.

经过这篇文章,基本上应该对Tornado web服务的工做有了一个初步的了解. 若是想继续深刻,请

直接看源码, 毕竟Tornado的源码是很是友好的,不少实现都有很高的参考.