tornado异步原理--异步事件

Posted 2021-01-17 madun

tornado异步原理

tornado有四类异步事件：立即事件，定时器异步事件，io异步事件，Future异步事件。

tornado 的ioloop管理所有的异步事件，并在适当的时机调用异步事件的回掉函数。

四类异步事件均在ioloop的start函数中调度。

立即事件：

场景：当前函数执行完后，下次ioloop调度时直接调度某函数
用法：ioloop.add_callback(callback, *args, **kwargs)
原理：立即事件全部存放在ioloop._callbacks中，IOLoop每次循环都会调用这些立即事件的回调函数

def start(self):
    while True:
        ncallbacks = len(self._callbacks) #self._callbacks用于存放所有的立即事件
        due_timeouts = [] 
        if self._timeouts: 
            now = self.time()
            while self._timeouts:
                if self._timeouts[0].callback is None: 
                    heapq.heappop(self._timeouts) 
                    self._cancellations -= 1
                elif self._timeouts[0].deadline <= now: 
                    due_timeouts.append(heapq.heappop(self._timeouts))
                else: 
                    break
        for i in range(ncallbacks):
            self._run_callback(self._callbacks.popleft()) #循环调用所有的立即事件的回调函数
        for timeout in due_timeouts:
            if timeout.callback is not None:
                self._run_callback(timeout.callback) 

        if self._callbacks: #如果在上面调用回调函数的过程中，又添加了新的立即事件，则将等待IO事件的时间设置为0，以便及时调用新的立即事件
            poll_timeout = 0.0
        elif self._timeouts:   
            poll_timeout = self._timeouts[0].deadline - self.time()
            poll_timeout = max(0, min(poll_timeout, _POLL_TIMEOUT))
        else:
            poll_timeout = _POLL_TIMEOUT

        event_pairs = self._impl.poll(poll_timeout)
        self._events.update(event_pairs)
        while self._events:
            fd, events = self._events.popitem()
            fd_obj, handler_func = self._handlers[fd]
            handler_func(fd_obj, events)

 

定时器异步事件：
场景：用户希望在某一段时间后执行某函数
用法：ioloop.call_at(when, callback, *args, **kwargs), ioloop.call_later(delay, callback, *args, **kwargs)
原理：定时器事件存放在ioloop._timeouts中，IOLoop每次循环开始都会找出所有已经超时的定时器，并调用对应的回调函数

def start(self):
    while True:
        ncallbacks = len(self._callbacks)
        due_timeouts = [] #用于存放超时的事件
        if self._timeouts: #self._timeouts用于存放所有定时器事件
            now = self.time()
            while self._timeouts:
                if self._timeouts[0].callback is None: #如果定时器事件没有回掉函数，则说明已经取消，直接丢弃
                    heapq.heappop(self._timeouts) #heapq是一个数据结构，它保证heapq[0]永远是最小的一个元素
                    self._cancellations -= 1
                elif self._timeouts[0].deadline <= now: #如果定时器已经超时，则取出并添加至due_timeouts中
                    due_timeouts.append(heapq.heappop(self._timeouts))
                else: #因为heapq的特性，如果执行到这一步，说明剩下事件都没有超时，退出循环
                    break
        for i in range(ncallbacks):
            self._run_callback(self._callbacks.popleft())
        for timeout in due_timeouts:
            if timeout.callback is not None:
                self._run_callback(timeout.callback) #循环调用所有已超时定时器事件的回调函数

        if self._callbacks:
            poll_timeout = 0.0
        elif self._timeouts:   #根据最小定时器事件的时间设置等待IO事件的时间
            poll_timeout = self._timeouts[0].deadline - self.time()
            poll_timeout = max(0, min(poll_timeout, _POLL_TIMEOUT))
        else:
            poll_timeout = _POLL_TIMEOUT

        event_pairs = self._impl.poll(poll_timeout)
        self._events.update(event_pairs)
        while self._events:
            fd, events = self._events.popitem()
            fd_obj, handler_func = self._handlers[fd]
            handler_func(fd_obj, events)

 

IO异步事件：
场景：等待某个文件描述符的某个事件，如TCPserver等待socket的READ事件
用法：ioloop.add_handler(fd, callback, events)
原理：所有的文件描述符全部存放在ioloop._impl中，windows平台下_impl是tornado.platform.select.SelectIOLoop对象
在linux平台下_impl是tornado.platform.epoll.EPollIOLoop对象，作用都是同时监听多个文件描述符

def start(self):
    while True:
        ncallbacks = len(self._callbacks) 
        due_timeouts = [] 
        if self._timeouts: 
            now = self.time()
            while self._timeouts:
                if self._timeouts[0].callback is None: 
                    heapq.heappop(self._timeouts) 
                    self._cancellations -= 1
                elif self._timeouts[0].deadline <= now: 
                    due_timeouts.append(heapq.heappop(self._timeouts))
                else: 
                    break
        for i in range(ncallbacks):
            self._run_callback(self._callbacks.popleft()) 
        for timeout in due_timeouts:
            if timeout.callback is not None:
                self._run_callback(timeout.callback) 

        if self._callbacks: 
            poll_timeout = 0.0
        elif self._timeouts:   
            poll_timeout = self._timeouts[0].deadline - self.time()
            poll_timeout = max(0, min(poll_timeout, _POLL_TIMEOUT))
        else:
            poll_timeout = _POLL_TIMEOUT

        event_pairs = self._impl.poll(poll_timeout) #监听所有文件描述符
        self._events.update(event_pairs) 
        while self._events:
            fd, events = self._events.popitem()
            fd_obj, handler_func = self._handlers[fd]
            handler_func(fd_obj, events)  #循环调用所有文件描述符对应的回调函数

 

Future异步事件：
场景：等待某个异步事件结束后执行回掉函数
用法：ioloop.add_future(future, callback)， future.add_done_callback(callback)
原理：异步事件结束后调用Future.set_result()，当执行set_result时将future所有的回掉函数添加为ioloop的立即事件

class Future(object):
    def set_result(self, result):
        self._result = result
        self._set_done()
        
    def _set_done(self):
        self._done = True
        if self._callbacks:
            from tornado.ioloop import IOLoop
            loop = IOLoop.current()
            for cb in self._callbacks:
                loop.add_callback(cb, self) #将所有的回掉函数设置为ioloop的立即事件
            self._callbacks = None