tornado异步非阻塞原理

怼烎@ 2021-11-05 16:10 300阅读 0赞

### 1.tornado框架核心代码分析（Snow类注释） ###

1．每个请求过来就会创建一个socket对象，并调用select去监听连接，select会将所有请求放到readable\_list列表中

2．使用while不断执行for循环遍历readable\_list，如果是新连接请求过来就加入inputs列表中

3．如果已经连接就调用self.process来获取请求头和请求体，如果已经获取到了正常的返回内容，就会返回一个

HttpResponse类型，直接返回response返回值即可

4．如果没有处理完成就会返回一个future对象，将这个future对象加入async\_request\_handler字典中，程序会继续向下走，不会阻塞

5．每次执行完for循环后就会调用self.polling\_callback()方法，在这个方法中再使用for循环遍历async\_request\_handler字典中的future对象，只要future对象有response,就将response返回，并将这个future对象从字典中删除

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

class Snow(object):
        def __init__(self, routes):
            self.routes = routes
            self.inputs = set()
            self.request = None
            self.async_request_handler = {}
    
        def run(self, host='localhost', port=9999):
            """
            事件循环
            :param host:
            :param port:
            """
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind((host, port,))
            sock.setblocking(False)
            sock.listen(128)
            sock.setblocking(0)
            self.inputs.add(sock)
            try:
                while True:
                    readable_list, writeable_list, error_list = select.select(self.inputs, [], self.inputs,0.005)
                    for conn in readable_list:
                        if sock == conn:                        #1.表示有新连接请求过来
                            client, address = conn.accept()     #接收请求对象
                            client.setblocking(False)
                            self.inputs.add(client)             #加入inputs中
                        else:                                   #2. 连上后执行这里并判断连接类型
                            gen = self.process(conn)            # 获取请求头请求体
                            if isinstance(gen, HttpResponse):   #2.1如果返回的HttpResponse类型，就直接返回response
                                    conn.sendall(gen.response())
                                    self.inputs.remove(conn)
                                conn.close()
                            else:                              #2.2如果返回的是一个future类型，加入字典，并hold住
                                yielded = next(gen)
                                # 将future对象放到字典中，hold住这个请求，就继续向下执行
                                self.async_request_handler[conn] = yielded
                    self.polling_callback()                    #3. 每次for循环结束就会调用这个方法
    
            except Exception as e:
                pass
            finally:
                sock.close()
    
        def polling_callback(self):
            """
            遍历触发异步非阻塞的回调函数
            :return:
            """
            for conn in list(self.async_request_handler.keys()):
                # conn是：   socket对象
                # yield是：  future对象
                yielded = self.async_request_handler[conn]
                # 若果future对象有返回值就会执行future.set_result()
                # 如果有人执行future.set_result()就会自动将ready改成true，才会向下走
                if not yielded.ready:
                    continue
                if yielded.callback:
                    ret = yielded.callback(self.request, yielded)
                    conn.sendall(ret.response())         #返回数据
                self.inputs.remove(conn)                 #将inputs中删除这个链接
                del self.async_request_handler[conn]    #在字典中删除这个future对象
                conn.close()
    
    
        def polling_callback(self):
            """
            4. 遍历触发异步非阻塞的回调函数，当future对象有返回就结束，并从字典中删除
            :return:
            """
            for conn in list(self.async_request_handler.keys()):
                # conn是：   socket对象
                # yield是：  future对象
                yielded = self.async_request_handler[conn]
                # 若果future对象有返回值就会执行future.set_result()
                # 如果有人执行future.set_result()就会自动将ready改成true，才会向下走
                if not yielded.ready:
                    continue
                if yielded.callback:
                    ret = yielded.callback(self.request, yielded)
                    conn.sendall(ret.response())  # 返回数据
                self.inputs.remove(conn)  # 将inputs中删除这个链接
                del self.async_request_handler[conn]  # 在字典中删除这个future对象
                conn.close()
    
        def process(self, conn):
            """
            处理路由系统以及执行函数
            :param conn:
            :return:
            """
            self.request = HttpRequest(conn)
            func = None
            for route in self.routes:
                if re.match(route[0], self.request.url):
                    func = route[1]
                    break
            if not func:
                return HttpNotFound()
            else:
                return func(self.request)

tornado核心处理类 Snow代码注释

### 2. 剖析Future()对象 实现异步非阻塞原理 ###

**1．原理说明**

1、当发送GET请求时，由于方法被@gen.coroutine装饰且yield 一个 Future对象，那么Tornado会等待

2、等待用户向future对象中放置数据或者发送信号，如果获取到数据或信号之后，就开始执行doing方法。

3、等待用户向future对象中放置数据或者发送信号，如果获取到数据或信号之后，就开始执行doing方法。

4、注意：在等待用户向future对象中放置数据或信号时，此连接是不断开的。

**2．验证方法**

1、首先访问：[http://127.0.0.1:8888/async][http_127.0.0.1_8888_async] 会阻塞，并且不断开，页面一直在转，说明非阻塞

2、然后访问：[http://127.0.0.1:8888/login][http_127.0.0.1_8888_login] 可以直接访问，证明可以实现异步

3、最后访问：[http://127.0.0.1:8888/stop][http_127.0.0.1_8888_stop] /index页面立刻就会返回了

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import tornado.ioloop
    import tornado.web
    from tornado import gen
    from tornado.concurrent import Future
    
    '''1. 访问：http://127.0.0.1:8888/async 会阻塞'''
    future = None
    
    class AsyncHandler(tornado.web.RequestHandler):
        @gen.coroutine
        def get(self):
            global future
            future = Future()
            future.add_done_callback(self.doing)
            yield future
    
            # from tornado import httpclient
            # http = httpclient.AsyncHTTPClient()
            # # 下载完成后，自动执行 future.set_result()
            # yield http.fetch("http://www.google.com", self.doing)
        def doing(self, *args, **kwargs):
            self.write('async')
            self.finish()
    
    '''2. 但是访问http://127.0.0.1:8888/login可以访问'''
    class LoginHandler(tornado.web.RequestHandler):
        def get(self):
            self.write('login')
    
    '''3. 当我们去访问http://127.0.0.1:8888/stop时就可以结束/async的阻塞'''
    class StopHandler(tornado.web.RequestHandler):
        def get(self, *args, **kwargs):
            future.set_result('...')
            self.write('结束阻塞')
    
    settings = {}
    application = tornado.web.Application([
        (r"/login", LoginHandler),
        (r"/async", AsyncHandler),
        (r"/stop", StopHandler),
    ],**settings)
    if __name__ == "__main__":
        application.listen(8888)
        print('直接访问会阻塞：http://127.0.0.1:8888/async')
        print('阻塞时还能访问login证明实现异步：http://127.0.0.1:8888/login')
        print('访问sotp会结束async的阻塞：http://127.0.0.1:8888/stop')
        tornado.ioloop.IOLoop.instance().start()

future对象实现阻塞 与 结束阻塞

转载于:https://www.cnblogs.com/junjun511/p/11298867.html

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211104/498a40aece27455bb6a00c01692ad411.png
[http_127.0.0.1_8888_async]: http://127.0.0.1:8888/async
[http_127.0.0.1_8888_login]: http://127.0.0.1:8888/login
[http_127.0.0.1_8888_stop]: http://127.0.0.1:8888/stop