사용자 정의된 Python 비동기 비차단 웹 프레임워크 예제 코드

이 기사에서는 200줄의 코드를 사용하여 소형 비동기 비차단 웹프레임워크를 완성합니다. Snow. 참고할만한 가치가 아주 좋습니다.

Python의 웹 프레임워크인 Tornado는 비동기 논블로킹으로 유명합니다. 이 기사에서는 200줄의 코드를 사용하여 소형 비동기 비차단 웹 프레임워크인 Snow를 완성합니다.

1. 소스 코드

이 기사는 비동기 비차단 웹 프레임워크를 구현하기 위한 비차단 소켓 및 IO 다중화를 기반으로 하며 그 중 많은 비동기 비차단이 있습니다. -웹 프레임워크 차단 프레임워크의 내부 원칙.

아아아아

2. 사용

1.

rreee

2. 비동기 비차단: 시간 초과

#!/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

3. 비동기 비차단: 대기대기 모드에 따라 맞춤형 작업 완료 가능

아아앙

위 내용은 사용자 정의된 Python 비동기 비차단 웹 프레임워크 예제 코드의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!