摘要：清楚了以上流程，我們直接來看函數(shù)主要用作初始化應用監(jiān)聽端口以及啟動。其中就是保存聊天室所有聊天消息的結構。關于的解讀我會放到閱讀源碼時講。然后把消息加到緩存里，如果緩存大于限制則取最新的條消息。

tornado 源碼自帶了豐富的 demo ，這篇文章主要分析 demo 中的聊天室應用： chatdemo

首先看 chatdemo 的目錄結構：

├── chatdemo.py
├── static
│?? ├── chat.css
│?? └── chat.js
└── templates
    ├── index.html
    ├── message.html
    └── room.html

非常簡單，基本沒有分層，三個模版一個 js 一個 css ，還有一個最重要的 chatdemo.py

本文的重點是弄清楚 chatdemo.py 的運行流程，所以對于此項目的其他文件，包括模版及 chat.js 的實現(xiàn)都不會分析，只要知道 chat.js 的工作流程相信對于理解 chatdemo.py 沒有任何問題

此 demo 主要基于長輪詢。 獲取新消息的原理：

在 chat.js 中有一個定時器會定時執(zhí)行 update 操作

當沒有新消息時 tornado 會一直 hold 住 chat.js 發(fā)來的 update 請求

當有新消息時 tornado 將包含新消息的數(shù)據(jù)返回給所有 hold 的 update 請求

此時 chat.js 收到 update 回復后更新返回數(shù)據(jù)在聊天室中，同時再進行一次 update 請求， 然后又從 1. 開始執(zhí)行。

發(fā)送新消息的原理：

輸入消息， 點擊 post 按鈕， chat.js 獲取表單后用 ajax 方式發(fā)送請求 new

tornado 收到請求 new ，返回消息本身， 同時通知所有 hold 住的 update 請求 ( 這里也包括發(fā)送 new 請求的 chat.js 所發(fā)送的 update 請求 ) 返回新消息

所有在線的 chat.js 收到 update 請求回復，更新返回信息到聊天室，同時再進行一次 update 請求。

清楚了以上流程，我們直接來看 chatdemo.py :

def main():
    parse_command_line()
    app = tornado.web.Application(
        [
            (r"/", MainHandler),
            (r"/a/message/new", MessageNewHandler),
            (r"/a/message/updates", MessageUpdatesHandler),
            ],
        cookie_secret="__TODO:_GENERATE_YOUR_OWN_RANDOM_VALUE_HERE__",
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path=os.path.join(os.path.dirname(__file__), "static"),
        xsrf_cookies=True,
        debug=options.debug,
        )
    app.listen(options.port)
    tornado.ioloop.IOLoop.current().start()


if __name__ == "__main__":
    main()

main 函數(shù)主要用作初始化應用、監(jiān)聽端口以及啟動 tornado server 。
我們看路由：

主頁對應 MainHandler

new 請求對應 MessageNewHandler

updates 請求對應 MessageUpdatesHandler

下面來看 MainHandler :

# Making this a non-singleton is left as an exercise for the reader.
global_message_buffer = MessageBuffer()

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("index.html", messages=global_message_buffer.cache)

只有一行代碼，就是渲染并返回 index.html，渲染的附加信息就是 global_message_buffer 的所有緩存消息。 global_message_buffer 是 MessageBuffer 的一個實例。 我們先不關心 MessageBuffer 內部是什么，現(xiàn)在我們只要記住它主要是用來儲存聊天消息和連接到此聊天室的人的信息的類。 其中 MessageBuffer().cache 就是保存聊天室所有聊天消息的結構。

然后來看 MessageNewHandler :

class MessageNewHandler(tornado.web.RequestHandler):
    def post(self):
        message = {
            "id": str(uuid.uuid4()),
            "body": self.get_argument("body"),
        }
        # to_basestring is necessary for Python 3"s json encoder,
        # which doesn"t accept byte strings.
        message["html"] = tornado.escape.to_basestring(
            self.render_string("message.html", message=message))
        if self.get_argument("next", None):
            self.redirect(self.get_argument("next"))
        else:
            self.write(message)
        global_message_buffer.new_messages([message])

同樣很簡單，從 post 信息里獲取發(fā)來的新消息 ( body ) ，然后給消息分配一個唯一的 uuid，接著把這段消息渲染成一段 html ，然后 self.write(message) 返回這段 html， 同時給 global_message_buffer ( MessageBuffer 的實例 ) 添加這條新信息。 這里其實我更傾向于返回 json 之類的數(shù)據(jù)，這樣更加直觀和規(guī)范，可能寫 demo 的人考慮到讀者對 json 之類的協(xié)議可能不熟悉故而選擇了返回渲染好的 html 直接讓 chat.js append 到 index.html 里。

接著來看 MessageUpdatesHandler :

class MessageUpdatesHandler(tornado.web.RequestHandler):
    @gen.coroutine
    def post(self):
        cursor = self.get_argument("cursor", None)
        # Save the future returned by wait_for_messages so we can cancel
        # it in wait_for_messages
        self.future = global_message_buffer.wait_for_messages(cursor=cursor)
        messages = yield self.future
        if self.request.connection.stream.closed():
            return
        self.write(dict(messages=messages))

    def on_connection_close(self):
        global_message_buffer.cancel_wait(self.future)

重點就在這里，可以看到其內部的 post 方法被 gen.coroutine 修飾器修飾，也就是說這個 post 方法現(xiàn)在是 協(xié)程 ( coroutine ) 方式工作。 對于協(xié)程比較陌生的童鞋，你可以直接把它當作是單線程解決 io ( 網(wǎng)絡請求 ) 密集運算被阻塞而導致低效率的解決方案。 當然這樣理解協(xié)程還比較籠統(tǒng)，之后我會詳細寫一篇關于協(xié)程的文章，但在這里這樣理解是沒有問題的。

現(xiàn)在來看代碼內容，首先獲取 cursor ，一個用來標識我們已經獲取的消息的指針，這樣 tornado 就不會把你已經獲取的消息重復的發(fā)給你。 然后調用 global_message_buffer.wait_for_messages(cursor=cursor) 獲取一個 future 對象。 future 對象是 tornado 實現(xiàn)的一個特殊的類的實例，它的作用就是包含之后 ( 未來 ) 將會返回的數(shù)據(jù)，我們現(xiàn)在不用關心 Future() 內部如何實現(xiàn)，只要記住上面它的作用就行。 關于 Future 的解讀我會放到閱讀 Future 源碼時講。

然后看最關鍵的這句： messages = yield self.future 注意這里的 yield 就是 hold updates 請求的關鍵，它到這里相當于暫停了整個 post 函數(shù) （ updates 請求被 hold ）同時也相當于 updates 這次網(wǎng)絡請求被阻塞，這個時候協(xié)程發(fā)揮作用，把這個函數(shù)暫停的地方的所有信息保存掛起，然后把工作線程釋放，這樣 tornado 可以繼續(xù)接受 new、 updates 等請求然后運行相應的方法處理請求。

當有新的消息返回時，tornado 底層的 ioloop 實例將會調用 gen.send(value) 返回新消息( value )給每個被暫停的方法的 yield 處， 此時協(xié)程依次恢復這些被暫停的方法， 同時用獲得的返回消息繼續(xù)執(zhí)行方法， 這時 messages = yield self.future 繼續(xù)執(zhí)行，messages 獲得 yield 的返回值 value ( python 中調用 gen.send(value) 將會把 value 值返回到 yield 處并替換原前 yield 后的值 )，然后判斷下用戶是否已經離開，如果還在線則返回新消息。

明白了以上流程，我們最后來看 MessageBuffer:

class MessageBuffer(object):
    def __init__(self):
        self.waiters = set()
        self.cache = []
        self.cache_size = 200

    def wait_for_messages(self, cursor=None):
        # Construct a Future to return to our caller.  This allows
        # wait_for_messages to be yielded from a coroutine even though
        # it is not a coroutine itself.  We will set the result of the
        # Future when results are available.
        result_future = Future()
        if cursor:
            new_count = 0
            for msg in reversed(self.cache):
                if msg["id"] == cursor:
                    break
                new_count += 1
            if new_count:
                result_future.set_result(self.cache[-new_count:])
                return result_future
        self.waiters.add(result_future)
        return result_future

    def cancel_wait(self, future):
        self.waiters.remove(future)
        # Set an empty result to unblock any coroutines waiting.
        future.set_result([])

    def new_messages(self, messages):
        logging.info("Sending new message to %r listeners", len(self.waiters))
        for future in self.waiters:
            future.set_result(messages)
        self.waiters = set()
        self.cache.extend(messages)
        if len(self.cache) > self.cache_size:
            self.cache = self.cache[-self.cache_size:]

初始化方法中 self.waiters 就是一個等待新消息的 listener 集合 ( 直接理解成所有被 hold 住的 updates 請求隊列可能更清晰 )

self.cache 就是儲存所有聊天消息的列表，self.cache_size = 200 則定義了 cache 的大小 是存 200 條消息。

然后先來看簡單的 new_messages:

遍歷 waiters 列表，然后給所有的等待者返回新消息，同時清空等待者隊列。 然后把消息加到緩存 cache 里，如果緩存大于限制則取最新的 200 條消息。這里只要注意到 future.set_result(messages) 就是用來給 future 對象添加返回數(shù)據(jù) ( 之前被 yield 暫停的地方此時因為 set_result() 方法將會獲得 "未來" 的數(shù)據(jù) ) 這一點即可。

然后來看 wait_for_messages :

    def wait_for_messages(self, cursor=None):
        # Construct a Future to return to our caller.  This allows
        # wait_for_messages to be yielded from a coroutine even though
        # it is not a coroutine itself.  We will set the result of the
        # Future when results are available.
        result_future = Future()
        if cursor:
            new_count = 0
            for msg in reversed(self.cache):
                if msg["id"] == cursor:
                    break
                new_count += 1
            if new_count:
                result_future.set_result(self.cache[-new_count:])
                return result_future
        self.waiters.add(result_future)
        return result_future

首先初始化一個 Future 對象，然后根據(jù) cursor 判斷哪些消息已經獲取了哪些還沒獲取，如果緩存中有對于這個 waiter 還沒獲取過的消息，則直接調用 set_result() 返回這些緩存中已有的但對于這個 waiter 來說是新的的數(shù)據(jù)。 如果這個 waiter 已經有緩存中的所有數(shù)據(jù)，那么就把它加到等待者隊列里保持等待，直到有新消息來時調用 new_messages 再返回。

而最后一個 cancel_wait 就很簡單了，當有用戶退出聊天室時，直接從 self.waiters 中移除他所對應的等待者。

當明白了整個代碼的運行流程后，我們可以基于這個簡單的 demo 而寫出更加豐富的例子，比如加入 session ，做登陸、做好友關系，做單聊做群聊等等。

chatdemo with room是我添加的一個簡單功能，輸入聊天室房號后再進行聊天，只有同一房間中的人才能收到彼此的消息。

以上就是鄙人對整個 chatdemo.py 的解讀。 在閱讀此 demo 時，我沒有參考其他源碼解讀，只是通過閱讀 tornado 底層的源碼而得出的個人的理解，因此肯定會有很多理解不成熟甚至錯誤的地方，還望大家多多指教。

原文地址

作者：rapospectre