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)
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,同时也希望多多支持!
# 异步web框架
# 异步非阻塞
# python
# python3中celery异步框架简单使用+守护进程方式启动
# python 5个顶级异步框架推荐
# python异步Web框架sanic的实现
# 关于Python核心框架tornado的异步协程的2种方法详解
# Python的Tornado框架实现异步非阻塞访问数据库的示例
# Python的Tornado框架的异步任务与AsyncHTTPClient
# Python的Twisted框架上手前所必须了解的异步编程思想
# Python的Tornado框架异步编程入门实例
# 简单介绍Python的Tornado框架中的协程异步实现原理
# 在Python的gevent框架下执行异步的Solr查询的教程
# python 常用的异步框架汇总整理
# 回调
# 遍历
# 多路
# 复用
# 错误提示
# 而闻名
# header_flag
# protocol
# url
# False
# True
# initialize
# initialize_headers
# conn
# header_bytes
# super
# HttpRequest
# body_bytes
# method
# header_dict
相关文章:
如何用wdcp快速搭建高效网站?
建站主机选哪种环境更利于SEO优化?
建站之星免费模板:自助建站系统与智能响应式一键生成
网站规划与制作是什么,电子商务网站系统规划的内容及步骤是什么?
网站按钮制作软件,如何实现网页中按钮的自动点击?
如何在Golang中实现微服务服务拆分_Golang微服务拆分与接口管理方法
建站主机CVM配置优化、SEO策略与性能提升指南
定制建站方案优化指南:企业官网开发与建站费用解析
制作国外网站的软件,国外有哪些比较优质的网站推荐?
如何快速使用云服务器搭建个人网站?
如何访问已购建站主机并解决登录问题?
大同网页,大同瑞慈医院官网?
广东企业建站网站优化与SEO营销核心策略指南
简历在线制作网站免费,免费下载个人简历的网站是哪些?
英语简历制作免费网站推荐,如何将简历翻译成英文?
婚礼视频制作网站,学习*后期制作的网站有哪些?
济南专业网站制作公司,济南信息工程学校怎么样?
建站之星IIS配置教程:代码生成技巧与站点搭建指南
建站之星如何保障用户数据免受黑客入侵?
厦门模型网站设计制作公司,厦门航空飞机模型掉色怎么办?
黑客如何通过漏洞一步步攻陷网站服务器?
清单制作人网站有哪些,近日“兴风作浪的姑奶奶”引起很多人的关注这是什么事情?
最好的网站制作公司,网购哪个网站口碑最好,推荐几个?谢谢?
如何用PHP快速搭建CMS系统?
家庭建站与云服务器建站,如何选择更优?
建站之星Pro快速搭建教程:模板选择与功能配置指南
弹幕视频网站制作教程下载,弹幕视频网站是什么意思?
电商网站制作多少钱一个,电子商务公司的网站制作费用计入什么科目?
,网站推广常用方法?
图册素材网站设计制作软件,图册的导出方式有几种?
如何挑选高效建站主机与优质域名?
如何配置IIS站点权限与局域网访问?
如何处理“XML格式不正确”错误 常见XML well-formed问题解决方法
建站之星价格显示格式升级,你的预算足够吗?
外贸公司网站制作哪家好,maersk船公司官网?
建站之星3.0如何解决常见操作问题?
如何配置支付宝与微信支付功能?
如何用手机制作网站和网页,手机移动端的网站能制作成中英双语的吗?
Swift中循环语句中的转移语句 break 和 continue
零基础网站服务器架设实战:轻量应用与域名解析配置指南
网站制作外包价格怎么算,招聘网站上写的“外包”是什么意思?
如何在宝塔面板中创建新站点?
免费视频制作网站,更新又快又好的免费电影网站?
如何在Windows虚拟主机上快速搭建网站?
如何在服务器上配置二级域名建站?
子杰智能建站系统|零代码开发与AI生成SEO优化指南
高防服务器租用如何选择配置与防御等级?
,想在网上投简历,哪几个网站比较好?
建站OpenVZ教程与优化策略:配置指南与性能提升
*请认真填写需求信息,我们会在24小时内与您取得联系。