tornado 是一個 python web 框架和異步網絡庫幢踏，使用 IO 事件循環(huán)
分析版本為：v3.0.0 瑰艘，先從早期的看起
- 事件循環(huán)
先簡要介紹一下事件循環(huán)的思想
loop = EventLoop() # 創(chuàng)建事件循環(huán)實例
loop.register_event(event) # 注冊事件
loop.add_handler(event, handler) # 注冊對應事件處理器
# 無限循環(huán)監(jiān)測事件
while True:
# 阻塞，等待事件發(fā)生或者超時
events = loop.poll(timeout)
# 尋找對應事件處理器框咙，處理事件跟畅，進入下一次循環(huán)
for e in events:
handler = loop.get_handler(event)
handler(e)
- helloworld
下面的代碼是 tornado 倉庫 demos/helloworld.py 代碼
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
from tornado.options import define, options
define("port", default=8888, help="run on the given port", type=int)
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
def main():
tornado.options.parse_command_line() # 1?? 命令行參數解析
application = tornado.web.Application([
(r"/", MainHandler),
]) # 2?? 初始化 application
http_server = tornado.httpserver.HTTPServer(application) # 3?? 初始化 httpserver
http_server.listen(options.port) # 4?? 監(jiān)聽端口
tornado.ioloop.IOLoop.instance().start() # 5?? 啟動 ioloop
if __name__ == "__main__":
main()
先從這個簡單的例子分析 tornado 的啟動過程
4?? 監(jiān)聽端口
http_server.listen(options.port)
這里實際是調用 tcpserver.py/TCPServer.listen 方法
# 文件 tcpserver.py
class TCPServer(object):
def listen(self, port, address=""):
sockets = bind_sockets(port, address=address)
self.add_sockets(sockets)
def add_sockets(self, sockets):
if self.io_loop is None:
self.io_loop = IOLoop.current()
for sock in sockets:
self._sockets[sock.fileno()] = sock
add_accept_handler(sock, self._handle_connection,
io_loop=self.io_loop)
# 文件 netutil.py
def add_accept_handler(sock, callback, io_loop=None):
if io_loop is None:
io_loop = IOLoop.current()
def accept_handler(fd, events):
while True:
try:
connection, address = sock.accept()
except socket.error as e:
if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
return
raise
callback(connection, address)
io_loop.add_handler(sock.fileno(), accept_handler, IOLoop.READ)
這個方法主要做了兩件事：1.創(chuàng)建 socket ，綁定 IP 端口堕扶；2.注冊 handler 到 IOLoop
5?? 啟動 ioloop
tornado.ioloop.IOLoop.instance().start()
ioloop 使用了單例模式，全局只有一個 IOLoop 實例
class IOLoop(Configurable):
@staticmethod
def instance():
"""Returns a global `IOLoop` instance.
Most applications have a single, global `IOLoop` running on the
main thread. Use this method to get this instance from
another thread. To get the current thread's `IOLoop`, use `current()`.
"""
if not hasattr(IOLoop, "_instance"):
with IOLoop._instance_lock:
if not hasattr(IOLoop, "_instance"):
# New instance after double check
IOLoop._instance = IOLoop()
return IOLoop._instance
針對不同的系統(tǒng)梭依，IOLoop 會選擇不同的實現（ platform 文件夾）稍算，以達到性能的最大化。
選擇的過程通過 python 的魔法方法 __new__ 實現
關于 python 的魔法方法的介紹可以看這篇文章
# 文件 util.py
class Configurable(object):
__impl_class = None
__impl_kwargs = None
def __new__(cls, **kwargs):
base = cls.configurable_base()
args = {}
if cls is base:
impl = cls.configured_class()
if base.__impl_kwargs:
args.update(base.__impl_kwargs)
else:
impl = cls
args.update(kwargs)
instance = super(Configurable, cls).__new__(impl)
# initialize vs __init__ chosen for compatiblity with AsyncHTTPClient
# singleton magic. If we get rid of that we can switch to __init__
# here too.
instance.initialize(**args)
return instance
Configurable.__new__ 調用了兩個類方法 configurable_base configured_class
在 helloworld 的例子中糊探，我們調用的是類 IOLoop 的方法，代碼如下
# 文件 ioloop.py
class IOLoop(Configurable):
@classmethod
def configurable_base(cls):
return IOLoop
@classmethod
def configurable_default(cls):
if hasattr(select, "epoll"):
from tornado.platform.epoll import EPollIOLoop
return EPollIOLoop
if hasattr(select, "kqueue"):
# Python 2.6+ on BSD or Mac
from tornado.platform.kqueue import KQueueIOLoop
return KQueueIOLoop
from tornado.platform.select import SelectIOLoop
return SelectIOLoop
接下來我們看看 start 的代碼（只呈現了主要邏輯）
# 文件 ioloop
class PollIOLoop(IOLoop):
def start(self):
while True:
poll_timeout = 3600.0
# Prevent IO event starvation by delaying new callbacks
# to the next iteration of the event loop.
with self._callback_lock:
callbacks = self._callbacks
self._callbacks = []
for callback in callbacks:
self._run_callback(callback)
if self._timeouts:
now = self.time()
while self._timeouts:
if self._timeouts[0].callback is None:
# the timeout was cancelled
heapq.heappop(self._timeouts)
elif self._timeouts[0].deadline <= now:
timeout = heapq.heappop(self._timeouts)
self._run_callback(timeout.callback)
else:
seconds = self._timeouts[0].deadline - now
poll_timeout = min(seconds, poll_timeout)
break
if self._callbacks:
# If any callbacks or timeouts called add_callback,
# we don't want to wait in poll() before we run them.
poll_timeout = 0.0
if not self._running:
break
try:
event_pairs = self._impl.poll(poll_timeout)
except Exception as e:
...
self._events.update(event_pairs)
while self._events:
fd, events = self._events.popitem()
self._handlers[fd](fd, events)
整個循環(huán)的過程可以分為：
? 1.執(zhí)行回調
? 2.執(zhí)行定時任務
? 3.監(jiān)測事件
? 4.處理發(fā)生的事件
? 5.進入下一次循環(huán)
