tornado异步原理--异步事件

Posted madun

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了tornado异步原理--异步事件相关的知识,希望对你有一定的参考价值。

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

 













以上是关于tornado异步原理--异步事件的主要内容,如果未能解决你的问题,请参考以下文章

tornado和异步IO的基本简单流程原理

tornado异步原理

Tornado学习之异步原理

tornado数据库操作怎么异步

Tornado异步非阻塞的使用以及原理

Tornado----自定义异步非阻塞Web框架:Snow