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 

相关栏目： 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571 】

相关推荐： Laravel中DTO是什么概念_在Laravel项目中使用数据传输对象(DTO)  Edge浏览器如何截图和滚动截图_微软Edge网页捕获功能使用教程【技巧】  Laravel如何设置定时任务（Cron Job）_Laravel调度器与任务计划配置  瓜子二手车官方网站在线入口 瓜子二手车网页版官网通道入口  独立制作一个网站多少钱,建立网站需要花多少钱？  Laravel如何生成和使用数据填充？（Seeder和Factory示例）  javascript中闭包概念与用法深入理解  Laravel如何处理异常和错误？（Handler示例）  如何撰写建站申请书？关键要点有哪些？  Laravel Artisan命令怎么自定义_创建自己的Laravel命令行工具完全指南  Win11摄像头无法使用怎么办_Win11相机隐私权限开启教程【详解】  百度浏览器如何管理插件 百度浏览器插件管理方法  如何将凡科建站内容保存为本地文件？  Laravel如何清理系统缓存命令_Laravel清除路由配置及视图缓存的方法【总结】  Laravel Docker环境搭建教程_Laravel Sail使用指南  Laravel如何配置任务调度？（Cron Job示例）  Laravel队列任务超时怎么办_Laravel Queue Timeout设置详解  Laravel如何使用.env文件管理环境变量？（最佳实践）  Laravel怎么配置.env环境变量_Laravel生产环境敏感数据保护与读取【方法】  Laravel辅助函数有哪些_Laravel Helpers常用助手函数大全  高性能网站服务器部署指南：稳定运行与安全配置优化方案  高防服务器租用首荐平台，企业级优惠套餐快速部署  Laravel中间件如何使用_Laravel自定义中间件实现权限控制  Laravel如何集成第三方登录_Laravel Socialite实现微信QQ微博登录  如何在服务器上配置二级域名建站？  HTML5空格在Angular项目里怎么处理_Angular中空格的渲染问题【详解】  JavaScript如何实现类型判断_typeof和instanceof有什么区别  北京的网站制作公司有哪些,哪个视频网站最好？  如何在万网开始建站？分步指南解析  如何在建站宝盒中设置产品搜索功能？  JavaScript如何操作视频_媒体API怎么控制播放  如何快速搭建FTP站点实现文件共享？  jQuery 常见小例汇总  Swift中switch语句区间和元组模式匹配  怎么用AI帮你设计一套个性化的手机App图标？  如何用西部建站助手快速创建专业网站？  手机网站制作平台,手机靓号代理商怎么制作属于自己的手机靓号网站？  JavaScript实现Fly Bird小游戏  如何在IIS中新建站点并配置端口与IP地址？  网站建设要注意的标准 促进网站用户好感度！  Laravel项目如何进行性能优化_Laravel应用性能分析与优化技巧大全  Laravel怎么实现模型属性的自动加密  Windows驱动无法加载错误解决方法_驱动签名验证失败处理步骤  javascript读取文本节点方法小结  如何快速登录WAP自助建站平台？  html5怎么画眼睛_HT5用Canvas或SVG画眼球瞳孔加JS控制动态【绘制】  悟空浏览器如何设置小说背景色_悟空浏览器背景色设置【方法】  Laravel如何使用Contracts（契约）进行编程_Laravel契约接口与依赖反转  如何解决hover在ie6中的兼容性问题  怎么制作网站设计模板图片,有电商商品详情页面的免费模板素材网站推荐吗？