本篇将使用200行代码完成一个微型异步非阻塞web框架：snow。具有很好的参考价值，下面跟着小编一起来看下吧
python的web框架中tornado以异步非阻塞而闻名。本篇将使用200行代码完成一个微型异步非阻塞web框架：snow。
一、源码
本文基于非阻塞的socket以及io多路复用从而实现异步非阻塞的web框架，其中便是众多异步非阻塞web框架内部原理。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import re
import socket
import select
import time
class httpresponse(object):
“””
封装响应信息
“””
def init(self, content=”):
self.content = content
self.headers = {}
self.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
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 snow(object):
“””
微型web框架类
“””
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:
:return:
“””
sock = socket.socket(socket.af_inet, socket.sock_stream)
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)
if isinstance(gen, httpresponse):
conn.sendall(gen.response())
self.inputs.remove(conn)
conn.close()
else:
yielded = next(gen)
self.async_request_handler[conn] = yielded
self.polling_callback()
except exception as e:
pass
finally:
sock.close()
def polling_callback(self):
“””
遍历触发异步非阻塞的回调函数
:return:
“””
for conn in list(self.async_request_handler.keys()):
yielded = self.async_request_handler[conn]
if not yielded.ready:
continue
if yielded.callback:
ret = yielded.callback(self.request, yielded)
conn.sendall(ret.response())
self.inputs.remove(conn)
del self.async_request_handler[conn]
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)
snow.py
二、使用
1. 基本使用
from snow import snow
from snow import httpresponse
def index(request):
return httpresponse(‘ok’)
routes = [
(r’/index/’, index),
]
app = snow(routes)
app.run(port=8012)
2.异步非阻塞：超时
from snow import snow
from snow import httpresponse
from snow import timeoutfuture
request_list = []
def async(request):
obj = timeoutfuture(5)
yield obj
def home(request):
return httpresponse(‘home’)
routes = [
(r’/home/’, home),
(r’/async/’, async),
]
app = snow(routes)
app.run(port=8012)
3.异步非阻塞：等待
基于等待模式可以完成自定制操作
from snow import snow
from snow import httpresponse
from snow import future
request_list = []
def callback(request, future):
return httpresponse(future.value)
def req(request):
obj = future(callback=callback)
request_list.append(obj)
yield obj
def stop(request):
obj = request_list[0]
del request_list[0]
obj.set_result(‘done’)
return httpresponse(‘stop’)
routes = [
(r’/req/’, req),
(r’/stop/’, stop),
]
app = snow(routes)
app.run(port=8012)
以上就是自定义python异步非阻塞web框架实例代码的详细内容，更多请关注 第一php社区 其它相关文章！