博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
自定制异步非阻塞web框架
阅读量:4961 次
发布时间:2019-06-12

本文共 6214 字,大约阅读时间需要 20 分钟。

  本框架是分析了Tornado源码之后,编写出的一个微型的异步非阻塞web框架。基本上实现了Tornado异步非阻塞底层原理。

  异步非阻塞内部原理剖析:

  正常情况下,服务端处在一个单线程事件循环的过程中。一旦有人链接进来,则将這个链接加入到select的监听对象,只要该队列中某一个链接发生变化,我们就能及时处理该请求链接。当一个链接进来,如果服务端函数处理的结果是一个HttpResponse对象,也就是说,這个请求,服务端已经处理完毕了,可以直接返回。那么服务端直接通conn.sendall()将响应的数据全部发送给客户端。如果服务端函数处理的结果是一个Future对象,这就给服务端发送一个信号意思是我這个请求还没有结束,你别给我断开链接了。此时服务端要完成以下事情:第一,将返回的Future对象放入一个队列中;第二,继续监听其它请求,也就是说,在這个地方,我服务端并不会阻塞等待处理完毕,而是直接进入到时间循环。等待新的链接。第三,在事件循环的过程中,除了要监听链接的变化之外,还得监听上面的阻塞队列的中变化。至于阻塞队列如何变化,這个得通过set_result()信号,一旦這个被设置了,说明我這个阻塞的链接是时候返回了。整个过程,只用了一个线程在监听,完成了异步非阻塞的强大功能。

class HttpResponse(object):    """    封装响应信息    """    def __init__(self, content=''):        self.content = content  # 传过来的所有信息        self.headers = {}  # 请求头信息        self.cookies = {}  # 请求cookies    def response(self):        return bytes(self.content, encoding='utf-8')
class HttpNotFound(HttpResponse):    """    404时的错误提示    """    def __init__(self):        super(HttpNotFound, self).__init__('404 Not Found')
class HttpRequest(object):    """    用户封装用户请求信息    """    def __init__(self, conn):        self.conn = conn        self.header_bytes = bytes()        self.header_dict = {}        self.body_bytes = bytes()        self.method = ""        self.url = ""        self.protocol = ""        self.initialize()  # 预留钩子        self.initialize_headers()     def initialize(self):        header_flag = False        while True:            try:                received = self.conn.recv(8096)            except Exception as e:                received = None            if not received:                break            if header_flag:                self.body_bytes += received                continue            temp = received.split(b'\r\n\r\n', 1)            if len(temp) == 1:                self.header_bytes += temp            else:                h, b = temp                self.header_bytes += h                self.body_bytes += b                header_flag = True    @property    def header_str(self):        return str(self.header_bytes, encoding='utf-8')    def initialize_headers(self):        headers = self.header_str.split('\r\n') # 分离出请求体与请求头        first_line = headers[0].split(' ') # 分离请求头        if len(first_line) == 3:            self.method, self.url, self.protocol = headers[0].split(' ')             for line in headers:                kv = line.split(':')                if len(kv) == 2:                    k, v = kv                    self.header_dict[k] = v
class Future(object):    """    异步非阻塞模式时封装回调函数以及是否准备就绪    """    def __init__(self, callback):        self.callback = callback        self._ready = False # 返回与否的标识,False不返回,True返回        self.value = None    def set_result(self, value=None):        self.value = value        self._ready = True    @property    def ready(self):        return self._ready
class TimeoutFuture(Future):    """    异步非阻塞超时    """    def __init__(self, timeout):        super(TimeoutFuture, self).__init__(callback=None)        self.timeout = timeout        self.start_time = time.time()    @property    def ready(self):        current_time = time.time()        if current_time > self.start_time + self.timeout:            self._ready = True        return self._ready
class Rain(object):    """    微型Web框架类    """    def __init__(self, routes):        self.routes = routes        self.inputs = set()        self.request = None        self.async_request_handler = {} # 存放等待返回的Future对象    def run(self, host='localhost', port=9999):        """        事件循环        :param host:        :param port:        :return:        """        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:                          client, address = conn.accept()                        client.setblocking(False)                        self.inputs.add(client)                    else:                        gen = self.process(conn) # 返回有两种可能:1.HttpResponse对象,2.生成器对象                        if isinstance(gen, HttpResponse): # 返回HttpResponse对象                            conn.sendall(gen.response())                            self.inputs.remove(conn)                            conn.close()                        else: # 返回生成器对象                            yielded = next(gen) # 取第一个yield                            self.async_request_handler[conn] = yielded #{conn:Future对象}                self.polling_callback() # 循环检测async_request_handler中等待返回的Future对象        except Exception as e:            pass        finally:            sock.close()    def polling_callback(self):        """        遍历触发异步非阻塞的回调函数        :return:        """        # 循环检测async_request_handler中等待返回的Future对象 ,一旦某个Future对象set_result()被设置了,该对象就应该返回了        for conn in list(self.async_request_handler.keys()):            yielded = self.async_request_handler[conn] # Future对象            if not yielded.ready:  # 判断ready是否被设置为True,如果 被设置为True,则应该执行回调函数                continue            if yielded.callback:                ret = yielded.callback(self.request, yielded)                conn.sendall(ret.response())            self.inputs.remove(conn)            del self.async_request_handler[conn] # Http应该返回了            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)

  使用方式(上述代码整合到一个文件,保存为:MyTornado):

from MyTornado import Rainfrom MyTornado import HttpResponsedef index(request):    return HttpResponse('OK')routes = [    (r'/index/', index),]app = Rain(routes)app.run(port=8012)

 

转载于:https://www.cnblogs.com/vipchenwei/articles/8046599.html

你可能感兴趣的文章
在使用Kettle的集群排序中 Carte的设定——(基于Windows)
查看>>
【原】iOS中KVC和KVO的区别
查看>>
OMAPL138学习----DSPLINK DEMO解析之SCALE
查看>>
IoC的基本概念
查看>>
restframework CBV试图的4种方式
查看>>
大图居中,以1920px为例
查看>>
Python3 图片转字符画
查看>>
[C陷阱和缺陷] 第7章 可移植性缺陷
查看>>
人需要治愈
查看>>
linux中configure文件默认执行结果所在位置
查看>>
Windows向Linux上传文件夹
查看>>
20180104-高级特性-Slice
查看>>
6个SQL Server 2005性能优化工具介绍
查看>>
nginx启动、关闭命令、重启nginx报错open() "/var/run/nginx/nginx.pid" failed
查看>>
BZOJ 3097 Hash Killer I
查看>>
UINavigationController的视图层理关系
查看>>
html阴影效果怎么做,css 内阴影怎么做
查看>>
宏观经济
查看>>
综合练习:词频统计
查看>>
BZOJ1026: [SCOI2009]windy数
查看>>