后端开发|Python教程
python Web框架
后端开发-Python教程
本篇将使用200行代码完成一个微型异步非阻塞Web框架：Snow。具有很好的参考价值，下面跟着小编一起来看下吧
pho源码,vscode自动跳转最下,ubuntu 下卡死,tomcat 防止盗链,sqlite数据库 文档,手机版爬虫抢茅台软件下载,php qq发邮件,佛山营销seo托管,精良网站源码,网页抽奖js,jquery登录界面模板lzw
Python的Web框架中Tornado以异步非阻塞而闻名。本篇将使用200行代码完成一个微型异步非阻塞Web框架：Snow。
网上蛋糕源码,怎么防止ubuntu死机,java创建爬虫项目,.php html,澳门seo价格lzw
A、源码
flash坦克大战源码,vscode去缩进,ubuntu怎么查询文件路径,tomcat固定网卡,sqlite重组微信,js预览pdf插件,前端处理图片js框架,用java做爬虫,论坛源码 php,南宁SEO优化电池,域名网站程序哪里有下载,网页源代码保护,茶叶模板 茶叶网站程序,简约的登录页面,客户管理系统破解版源码,网址导航模板程序lzw
本文基于非阻塞的Socket以及IO多路复用从而实现异步非阻塞的Web框架，其中便是众多异步非阻塞Web框架内部原理。
#!/usr/bin/env python# -*- coding:utf-8 -*-import reimport socketimport selectimport timeclass 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\ \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(\ \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 = kvself.header_dict[k] = vclass 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._readyclass 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._readyclass 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
B、使用
1. 基本使用
from snow import Snowfrom snow import HttpResponsedef index(request):return HttpResponse(OK)routes = [ (r/index/, index),]app = Snow(routes)app.run(port=8012)
2.异步非阻塞：超时
from snow import Snowfrom snow import HttpResponsefrom snow import TimeoutFuturerequest_list = []def async(request): obj = TimeoutFuture(5) yield objdef home(request): return HttpResponse(home)routes = [ (r/home/, home), (r/async/, async),]app = Snow(routes)app.run(port=8012)
3.异步非阻塞：等待
基于等待模式可以完成自定制操作
from snow import Snowfrom snow import HttpResponsefrom snow import Futurerequest_list = []def callback(request, future): return HttpResponse(future.value)def req(request): obj = Future(callback=callback) request_list.append(obj) yield objdef 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框架实例代码》对你有帮助，请点赞、收藏，并留下你的观点哦！
