（伯乐在线转注：本文英文原文写于 2009 年，赵杰翻译于 2011 年）

• 自己写一个Comet架构程序的服务器端部分，即使你是从拷贝别人的代码开始。
• 如果你想在Tornado框架上做开发，通过这篇文章你将更好的理解Tornado web框架。
• 在Tornado和Twisted的争论上，你将更有见解。

介绍

def handler_request(self, request):
    answ = self.remote_server.query(request) # this takes 5 seconds
    request.write_response(answ)

def handler_request(self, request):
    self.remote_server.query_async(request, self.response_received)     
def response_received(self, request, answ):    # this is called 5 seconds later
    request.write(answ)

git clone git://github.com/facebook/tornado.git

def add_handler(self, fd, handler, events):
    """Registers the given handler to receive the given events for fd."""
    self._handlers[fd] = handler
    self._impl.register(fd, events | self.ERROR)

def start(self):
    """Starts the I/O loop.
    The loop will run until one of the I/O handlers calls stop(), which
    will make the loop stop after the current event iteration completes.
    """
    self._running = True
    while True:
    [ ... ]
        if not self._running:
            break
        [ ... ]
        try:
            event_pairs = self._impl.poll(poll_timeout)
        except Exception, e:
            if e.args == (4, "Interrupted system call"):
                logging.warning("Interrupted system call", exc_info=1)
                continue
            else:
                raise
        # Pop one fd at a time from the set of pending fds and run
        # its handler. Since that handler may perform actions on
        # other file descriptors, there may be reentrant calls to
        # this IOLoop that update self._events
        self._events.update(event_pairs)
        while self._events:
            fd, events = self._events.popitem()
            try:
                self._handlers[fd](fd, events)
            except KeyboardInterrupt:
                raise
            except OSError, e:
                if e[0] == errno.EPIPE:
                    # Happens when the client closes the connection
                    pass
                else:
                    logging.error("Exception in I/O handler for fd %d",
                                  fd, exc_info=True)
            except:
                logging.error("Exception in I/O handler for fd %d",
                              fd, exc_info=True)

for fd, events in self._events.items():

def __init__(self, impl=None):
    [...]
    # Create a pipe that we send bogus data to when we want to wake
    # the I/O loop when it is idle
    r, w = os.pipe()
    self._set_nonblocking(r)
    self._set_nonblocking(w)
    self._waker_reader = os.fdopen(r, "r", 0)
    self._waker_writer = os.fdopen(w, "w", 0)
    self.add_handler(r, self._read_waker, self.WRITE)
def _wake(self):
    try:
        self._waker_writer.write("x")
    except IOError:
        pass

def add_timeout(self, deadline, callback):
    """Calls the given callback at the time deadline from the I/O loop."""
    timeout = _Timeout(deadline, callback)
    bisect.insort(self._timeouts, timeout)
    return timeout

# Choose a poll implementation. Use epoll if it is available, fall back to
# select() for non-Linux platforms
if hasattr(select, "epoll"):
    # Python 2.6+ on Linux
    _poll = select.epoll
else:
    try:
        # Linux systems with our C module installed
        import epoll
        _poll = _EPoll
    except:
        # All other systems
        import sys
        if "linux" in sys.platform:
            logging.warning("epoll module not found; using select()")
        _poll = _Select

• read_until()，从socket中读取直到遇到指定的字符串。这为在读取HTTP头时遇到空行分隔符自动停止提供了方便。
• read_bytes()，从socket中读取指定数量的字节。这为读取HTTP消息的body部分提供了方便。
• write()，将指定的buffer写入socket并持续监测直到这个buffer被发送。

所有上述的方法都可以通过异步方式在它们完成时触发回调函数。

def write(self, data, callback=None):
    """Write the given data to this stream.
    If callback is given, we call it when all of the buffered write
    data has been successfully written to the stream. If there was
    previously buffered write data and an old write callback, that
    callback is simply overwritten with this new callback.
    """
    self._check_closed()
    self._write_buffer += data
    self._add_io_state(self.io_loop.WRITE)
    self._write_callback = callback

def _handle_read(self):
    try:
        chunk = self.socket.recv(self.read_chunk_size)
    except socket.error, e:
        if e[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
            return
        else:
            logging.warning("Read error on %d: %s",
                            self.socket.fileno(), e)
            self.close()
            return
    if not chunk:
        self.close()
        return
    self._read_buffer += chunk
    if len(self._read_buffer) >= self.max_buffer_size:
        logging.error("Reached maximum read buffer size")
        self.close()
        return
    if self._read_bytes:
        if len(self._read_buffer) >= self._read_bytes:
            num_bytes = self._read_bytes
            callback = self._read_callback
            self._read_callback = None
            self._read_bytes = None
            callback(self._consume(num_bytes))
    elif self._read_delimiter:
        loc = self._read_buffer.find(self._read_delimiter)
        if loc != -1:
            callback = self._read_callback
            delimiter_len = len(self._read_delimiter)
            self._read_callback = None
            self._read_delimiter = None
            callback(self._consume(loc + delimiter_len))

def _consume(self, loc):
    result = self._read_buffer[:loc]
    self._read_buffer = self._read_buffer[loc:]
    return result

def listen(self, port, address=""):
    assert not self._socket
    self._socket = socket.(socket.AF_INET, socket.SOCK_STREAM, 0)
    flags = fcntl.fcntl(self._socket.fileno(), fcntl.F_GETFD)
    flags |= fcntl.FD_CLOEXEC
    fcntl.fcntl(self._socket.fileno(), fcntl.F_SETFD, flags)
    self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self._socket.setblocking(0)
    self._socket.bind((address, port))
    self._socket.listen(128)
    self.io_loop.add_handler(self._socket.fileno(), self._handle_events,
                             self.io_loop.READ)

除了绑定给定的地址和端口外，上述代码还设置了”close on exec”和”reuse address”这两个标志位。前者在应用程序创建子进程的时候特别有用。在这种情况下，我们不想让套接字保持打开的状态（任何设置了”close on exec”标志位的文件描述符，都不能被使用exec函数方式创建的子进程读写，因为该文件描述符在exec函数调用前就会被自动释放，译者注）。后者用来避免在服务器重启的时候发生“该地址以被使用”这种错误时很有用。

正如你所见到的，后备连接所允许的最大数目是128（注意，listen方法并不是你想象中的“开始在128端口上监听”的意思，译者注）。这意味着如果有128个连接正在等待被accept，那么直到服务器有时间将前面128个连接中的某几个accept了，新的连接都将被拒绝。我建议你在做性能测试的时候将该参数调高，因为当新的连接被抛弃的时候将直接影响你做测试的准确性。

在上述代码中注册的_handle_events()处理器用来accept新连接，并创建相关的IOStream对象和初始化一个HTTPConnection对象，HTTPConnection对象负责处理剩下的交互部分：

def _handle_events(self, fd, events):
    while True:
        try:
            connection, address = self._socket.accept()
        except socket.error, e:
            if e[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
                return
            raise
        try:
            stream = iostream.IOStream(connection, io_loop=self.io_loop)
            HTTPConnection(stream, address, self.request_callback,
                           self.no_keep_alive, self.xheaders)
        except:
            logging.error("Error in connection callback", exc_info=True)

HTTP头的部分的解析工作开始于HTTPConnection类的构造函数__init()__()：

def __init__(self, stream, address, request_callback, no_keep_alive=False,
             xheaders=False):
    self.stream = stream
    self.address = address
    self.request_callback = request_callback
    self.no_keep_alive = no_keep_alive
    self.xheaders = xheaders
    self._request = None
    self._request_finished = False
    self.stream.read_until("rnrn", self._on_headers)

_on_headers()回调函数实际用来解析HTTP头，并在有请求内容的情况下通过使用read_bytes()来读取请求的内容部分。_on_request_body()回调函数用来解析POST的参数并调用应用层提供的回调函数：

def _on_headers(self, data):
    eol = data.find("rn")
    start_line = data[:eol]
    method, uri, version = start_line.split(" ")
    if not version.startswith("HTTP/"):
        raise Exception("Malformed HTTP version in HTTP Request-Line")
    headers = HTTPHeaders.parse(data[eol:])
    self._request = HTTPRequest(
        connection=self, method=method, uri=uri, version=version,
        headers=headers, remote_ip=self.address[0])
    content_length = headers.get("Content-Length")
    if content_length:
        content_length = int(content_length)
        if content_length > self.stream.max_buffer_size:
            raise Exception("Content-Length too long")
        if headers.get("Expect") == "100-continue":
            self.stream.write("HTTP/1.1 100 (Continue)rnrn")
        self.stream.read_bytes(content_length, self._on_request_body)
        return
    self.request_callback(self._request)
def _on_request_body(self, data):
    self._request.body = data
    content_type = self._request.headers.get("Content-Type", "")
    if self._request.method == "POST":
        if content_type.startswith("application/x-www-form-urlencoded"):
            arguments = cgi.parse_qs(self._request.body)
            for name, values in arguments.iteritems():
                values = [v for v in values if v]
                if values:
                    self._request.arguments.setdefault(name, []).extend(
                        values)
        elif content_type.startswith("multipart/form-data"):
            boundary = content_type[30:]
            if boundary: self._parse_mime_body(boundary, data)
    self.request_callback(self._request)

def write(self, chunk):
    assert self._request, "Request closed"
    self.stream.write(chunk, self._on_write_complete)

未完待续？

通过这篇文章，我已经涵盖了从socket到应用层的所有方面。这应该能给你关于Tornado是如何工作的一个清晰的理解。总之，我认为Tornado的代码是非常友好的，我希望你也这样认为。

Tornado框架还有很大一部分我们没有探索，比如wep.py（应该是web.py，译者注）这个实际与你应用打交道的模块，又或者是template engine模块。如果我有足够兴趣的话，我也会介绍这些部分。可以通过订阅我的RSS feed来鼓励我。