一. Tornado是什麼?

Facebook釋出了開源網路伺服器框架Tornado，該平臺基於Facebook剛剛收購的社交聚合網站FriendFeed的實時資訊服務開發而來.Tornado由Python編寫，是一款輕量級的Web伺服器，同時又是一個開發框架。採用非阻塞I/O模型(epoll)，主要是為了應對高併發 訪問量而被開發出來，尤其適用於comet應用。

二. 為什麼要閱讀Tornado的原始碼 Tornado由前google員工開發, 程式碼非常精練, 實現也很輕巧, 加上清晰的註釋和豐富的demo, 我們可以很容易的閱讀分析tornado. 通過閱讀Tornado的原始碼, 你將學到:    * 理解Tornado的內部實現, 使用tornado進行web開發將更加得心應手     * 如何實現一個高效能,非阻塞的http伺服器     * 如何實現一個web框架     * 各種網路程式設計的知識, 比如epoll     * python程式設計的絕佳實踐 三. 從http伺服器開始

Tornado不僅是一個web開發框架, 還自己實現了一個http伺服器. 談到http伺服器, 我們自然想到C10K. 其中介紹了很多種伺服器的程式設計模型, tornado的http伺服器採用的是:  多程序 + 非阻塞 + epoll + pre-fork 模型 在分析tornado伺服器之前, 有必要了解web伺服器的工作流程. 四 http伺服器工作三部曲 從實現上來說, web伺服器是這樣工作的: (1) 建立listen socket, 在指定的監聽埠, 等待客戶端請求的到來 (2) listen socket接受客戶端的請求, 得到client socket, 接下來通過client socket與客戶端通訊 (3) 處理客戶端的請求, 首先從client socket讀取http請求的協議頭, 如果是post協議, 還可能要       讀取客戶端上傳的資料, 然後處理請求, 準備好客戶端需要的資料, 通過client socket寫給客戶端 五 Hello World from Http Server

為了更加理解web伺服器的工作流程, 我們使用python編寫一個簡單的http伺服器, 返回Hello, World給瀏覽器 Python程式碼  

1. import socket  
2. def handle_request(client):  
3.   buf = client.recv(1024)  
4.   print buf  
5.   client.send("HTTP/1.1 200 OK\r\n\r\n")  
6.   client.send("Hello, World")  
7. def main():  
8.   sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
9.   sock.bind(('localhost',8080))  
10.   sock.listen(5)  
11.   while True:  
12.     connection, address = sock.accept()  
13.     handle_request(connection)  
14.     connection.close()  
15. if __name__ == '__main__':  
16.   main()  

 

執行如下:

六. Hello World from Tornado Http Server

Tornado不能算是一個完整的http伺服器, 它只實現小部分的http協議, 大部分要靠使用者去實現.

tornado其實是一個伺服器開發框架, 使用它我們可以快速的開發一個高效的http伺服器. 下面我們

就使用tornado再寫一個Hello, World的Http伺服器.

Python程式碼  

1. #!/usr/bin/env python  
2. # -*- coding:utf-8 -*-  
3. import tornado.httpserver  
4. import tornado.ioloop  
5. def handle_request(request):  
6.    message = "Hello World from Tornado Http Server"  
7.    request.write("HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n%s" % (  
8.                  len(message), message))  
9.    request.finish()  
10. http_server = tornado.httpserver.HTTPServer(handle_request)  
11. http_server.listen(8080)  
12. tornado.ioloop.IOLoop.instance().start()  

 執行如下:

實現非常簡單, 只需要定義自己的處理方法, 其它的東西全部交給Tornado完成. 簡單看一下Tornado做了哪些工作.

首先建立HTTPServer類, 並把我們的處理方法傳遞過去

然後在8080開始監聽

最後啟動事件迴圈, 開始監聽網路事件. 主要是socket的讀和寫

到了這裡, 我有點等不及了, 迫切想了解tornado的內部實現是怎麼樣的. 特別是想知道Tornado的IOLoop到底是如何

工作的. 接下來我們開始解剖Tornado

七. Tornado伺服器概覽

理解了web伺服器的工作流程之後, 我們再來看看Tornado伺服器是如何實現這些處理流程的.

Tornado伺服器有3大核心模組:

(1) IOLoop

與我們上面那個簡陋的http伺服器不同, Tornado為了實現高併發和高效能, 使用了一個

IOLoop來處理socket的讀寫事件, IOLoop基於epoll, 可以高效的響應網路事件. 這是Tornado

高效的保證. 

(2) IOStream

為了在處理請求的時候, 實現對socket的非同步讀寫, Tornado實現了IOStream類, 用來處理socket

的非同步讀寫. 

(3) HTTPConnection

這個類用來處理http的請求, 包括讀取http請求頭, 讀取post過來的資料, 呼叫使用者自定義的處理方法,

以及把響應資料寫給客戶端socket

下面這幅圖描述了tornado伺服器的大體處理流程, 接下來我們將會詳細分析每一步流程的實現

八. 建立listen socket

httpserver.py, 定位到bind方法:

Python程式碼  

1. for res in socket.getaddrinfo(address, port, family, socket.SOCK_STREAM,  
2.                             0, socket.AI_PASSIVE | socket.AI_ADDRCONFIG):  
3.   af, socktype, proto, canonname, sockaddr = res  
4.   # 建立listen socket  
5.   sock = socket.socket(af, socktype, proto)  
6.   # 設定socket的屬性   
7.   flags = fcntl.fcntl(sock.fileno(), fcntl.F_GETFD)  
8.   flags |= fcntl.FD_CLOEXEC  
9.   fcntl.fcntl(sock.fileno(), fcntl.F_SETFD, flags)  
10.   sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  
11.   if af == socket.AF_INET6:  
12.       if hasattr(socket, "IPPROTO_IPV6"):  
13.           sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)  
14.   sock.setblocking(0)  
15.   # bind 和 listen  
16.   sock.bind(sockaddr)  
17.   sock.listen(128)  
18.   # 加入ioloop  
19.   self._sockets[sock.fileno()] = sock  
20.   if self._started:  
21.       self.io_loop.add_handler(sock.fileno(), self._handle_events,  
22.                                ioloop.IOLoop.READ)<span style="white-space: normal;">  
23. </span>  

這是實現web伺服器的標準步驟, 首先getaddrinfo返回伺服器的所有網絡卡資訊, 每塊網絡卡上都要建立監聽客戶端的請求.

按照socket -> bind -> listen步驟走下來, 最後把新建的listen socket加入ioloop. 那麼ioloop又是個什麼東西呢?

暫時我們把ioloop理解為一個事件容器. 使用者把socket和回撥函式註冊到容器中, 容器內部會輪詢socket, 一旦某個socket

可以讀寫, 就呼叫回撥函式來處理socket的讀寫事件.

這裡, 我們只監聽listen socket的讀事件, 回撥函式為_handle_events, 一旦listen socket可讀, 說明客戶端請求到來, 

然後呼叫_handle_events接受客戶端的請求. 

九. accept

httpserver.py, 定位到_handle_events. 這個方法接受客戶端的請求. 

為了便於分析, 我把處理ssl那部分程式碼剝離出去了.

Python程式碼  

1. def _handle_events(self, fd, events):  
2.   while True:  
3.       try:  
4.           connection, address = self._sockets[fd].accept()  
5.       except socket.error, e:  
6.           if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):  
7.               return  
8.           raise  
9.       try:  
10.           stream = iostream.IOStream(connection, io_loop=self.io_loop)  
11.           HTTPConnection(stream, address, self.request_callback,  
12.                          self.no_keep_alive, self.xheaders)  
13.       except:  
14.           logging.error("Error in connection callback", exc_info=True)  

accept方法返回客戶端的socket(注意connection的型別是socket), 以及客戶端的地址

然後建立IOStream物件, 用來處理socket的非同步讀寫. 這一步會呼叫ioloop.add_handler把client socket加入ioloop

再然後建立HTTPConnection, 處理使用者的請求.

十. 建立IOStream

10.1 何為IOStream

accept完成後, 我們就可以用client socket與客戶端通訊了. 為了實現對client socket的非同步讀寫, 我們為client socket

建立兩個緩衝區: _read_buffer和_write_buffer, 寫: 先寫到_write_buffer, 讀: 從_read_buffer讀. 這樣我們就不用

直接讀寫socket, 進而實現非同步讀寫. 這些操作都封裝在IOStream類中, 概括來說,

IOStream對socket的讀寫做了一層封裝, 通過使用兩個緩衝區, 實現對socket的非同步讀寫.

10.2 IOStream的初始化

IOStream與socket是一一對應的, 初始化主要做4個工作

(1) 初始化IOStream對應的socket

(2) 分配輸入緩衝區_write_buffer

(3) 分配輸出緩衝區_read_buffer

(4) 把socket加入ioloop, 這樣當socket可讀寫的時候, 呼叫回撥函式_handle_events把資料從socket讀入buffer, 

     或者把資料從buffer傳送給socket

找到iosteram.py, 定位到__init__方法

Python程式碼  

1. self.socket = socket  
2. self.io_loop = io_loop or ioloop.IOLoop.instance()  
3. self._read_buffer = collections.deque()  
4. self._write_buffer = collections.deque()  
5. self.io_loop.add_handler(  
6.     self.socket.fileno(), self._handle_events, self._state)  

10.3 IOStream提供的介面

IOStream對外提供了3個介面, 用來對socket的讀寫

(1) write(data)

把資料寫入IOStream的_write_buffer

(2) read_until(delimiter, callback)

從_read_buffer讀取資料, delimiter作為讀取結束符, 完了呼叫callback

(3) read_bytes(num_of_bytes, callback)

從_read_buffer讀取指定大小的資料, 完了呼叫callback

read_until和read_bytes都會呼叫_read_from_buffer把從buffer讀取資料, 然後呼叫_consume消耗掉buffer中

的資料.

10.4 體驗非同步IO

下面我們來看一個非同步IO的例項, 這是一個非同步http client的例子, 使用IOStream來下載http://nginx.net/index.html

Python程式碼  

1. #!/usr/bin/env python  
2. # -*- coding:utf-8 -*-  
3. from tornado import ioloop  
4. from tornado import iostream  
5. import socket  
6. def send_request():  
7.     stream.write("GET /index.html HTTP/1.0\r\nHost: nginx.net\r\n\r\n")  
8.     stream.read_until("\r\n\r\n", on_headers)  
9. def on_headers(data):  
10.     headers = {}  
11.     for line in data.split("\r\n"):  
12.        parts = line.split(":")  
13.        if len(parts) == 2:  
14.            headers[parts[0].strip()] = parts[1].strip()  
15.     stream.read_bytes(int(headers["Content-Length"]), on_body)  
16. def on_body(data):  
17.     print data  
18.     stream.close()  
19.     ioloop.IOLoop.instance().stop()  
20. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)  
21. stream = iostream.IOStream(s)  
22. stream.connect(("nginx.net", 80), send_request)  
23. ioloop.IOLoop.instance().start()  

首先呼叫connect連線伺服器, 完成後回撥send_request發出請求, 並讀取伺服器返回的http協議頭, 然後回撥

on_headers解析協議頭, 然後呼叫read_bytes讀取資料體, 然後回撥on_body把資料打印出來. 最後關閉stream

可以看到, 這一系列的呼叫都是通過回撥函式實現的, 這就是非同步的處理方式.

10.5 IOStream響應ioloop事件

上面提到, IOStream初始化的時候, 把socket加入ioloop, 一旦socket可讀寫, 就呼叫回撥函式_handle_events處理IO

事件. 開啟iostream.py, 定位到_handle_events

Python程式碼  

1. def _handle_events(self, fd, events):  
2.     if not self.socket:  
3.         logging.warning("Got events for closed stream %d", fd)  
4.         return  
5.     try:  
6.         if events & self.io_loop.READ:  
7.             self._handle_read()  
8.         if not self.socket:  
9.             return  
10.         if events & self.io_loop.WRITE:  
11.             if self._connecting:  
12.                 self._handle_connect()  
13.             self._handle_write()  
14.         if not self.socket:  
15.             return  
16.         if events & self.io_loop.ERROR:  
17.             # We may have queued up a user callback in _handle_read or  
18.             # _handle_write, so don't close the IOStream until those  
19.             # callbacks have had a chance to run.  
20.             self.io_loop.add_callback(self.close)  
21.             return  
22.         state = self.io_loop.ERROR  
23.         if self.reading():  
24.             state |= self.io_loop.READ  
25.         if self.writing():  
26.             state |= self.io_loop.WRITE  
27.         if state != self._state:  
28.             self._state = state  
29.             self.io_loop.update_handler(self.socket.fileno(), self._state)  
30.     except:  
31.         logging.error("Uncaught exception, closing connection.",  
32.                       exc_info=True)  
33.         self.close()  
34.         raise  

 可以看到_handle_events根據IO事件的型別, 來呼叫不同的處理函式, 對於可讀事件, 呼叫handle_read來處理.

handle_read會從socket讀取資料, 然後把資料存到_read_buffer.