通过前面几个小节的学习，现在我们想要把之前学到的知识点给串联起来，实现一个很小型的Web框架。虽然很小，但是用到的知识点都是比较多的。如Socket编程，装饰器传参在实际项目中如何使用。通过这一节的学习，希望能把我们以前的知识点掌握的更加牢靠！
一、客户端与服务器通信过程
在这里我们以浏览器来说明访问服务器时，会做什么处理！
说明：
TCP服务器：
1. 服务器主要是用来处理客户端的连接请求，然后把请求的url，传递给框架来处理具体的逻辑；
2. 服务器中需要定义一个处理响应头和状态码的函数，并作为参数传递给web框架的接口；
3. 服务器与web框架之间的通信主要是通过WSGI协议提供的接口，这样他们各司其职，耦合度很低
web框架：
在框架中需要解决的难题：
如何把url与相应的函数建立起关联？
浏览器发送不同的请求需要被不同的函数截取，并返回响应体。通过装饰器传递参数的方式就可以把url与函数建立对应关系。装饰器中的参数就是用来匹配url的。
二、代码实现
服务器端：
import logging
import socket
import select
import re
import wsgi_web
logging.basicConfig(level=logging.WARNING,
filename='./web框架日志.txt',
filemode='w',
format='%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s')
class WebServer:
"""创建一个Web服务器"""
def __init__(self):
# 1.创建tcp socket
self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 2.设置地址可重用
self.tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 3.绑定到某个端口
self.tcp_socket.bind(('', 6060))
# 4.设置监听队列
self.tcp_socket.listen(128)
def run(self):
"""运行服务器"""
# 1.设置tcp_socket为非阻塞状态
self.tcp_socket.setblocking(False)
# 2.创建epoll对象，并注册服务器的监听事件
epoll = select.epoll()
epoll.register(self.tcp_socket.fileno(), select.EPOLLIN)
client_dict = dict()
# 3.不断遍历epoll列表，检查fd上有无事件发生
while True:
epoll_list = epoll.poll()
for fd, event in epoll_list:
if fd == self.tcp_socket.fileno():
# 有客户端来连接服务器
client_socket, client_addr = self.tcp_socket.accept()
# 注册客户端的事件
epoll.register(client_socket.fileno(), select.EPOLLIN)
# 客户端与其fd要建立关联
client_dict[client_socket.fileno()] = client_socket
else:
# 说明客户端发送数据过来
# 通过fd来处理客户端的请求
self.response_client(fd, client_dict, epoll)
def response_client(self, client_fd, client_dict, epoll):
"""处理客户端的请求"""
try:
req_heads = client_dict[client_fd].recv(1024).decode('utf-8')
except Exception as e:
logging.warning(e)
if not req_heads:
# 1.关闭客户端
client_dict[client_fd].close()
# 2.从监听队列中移除该客户端
client_dict.popitem()
# 3.取消该客户端的注册事件
epoll.unregister(client_fd)
try:
# print(req_heads.splitlines()[0])
# 1.解析客户端的请求url
match = re.match(r'[^/]+(/[^ ]*)', req_heads.splitlines()[0])
if match:
# 匹配成功
filename = match.group(1)
if filename == '/':
filename = '/index.html'
except Exception as e:
logging.warning(e)
# print('匹配数据出现了错误:{}'.format(e))
# 2.根据文件名去动态加载，然后伪装成静态页面发送
if filename.endswith('.html'):
url_params = dict()
# 1.使用WSGI接口
# 通过不同文件构造一个字典
url_params['filename'] = filename
# print(url_params)
# 定义一个函数传给框架
body = wsgi_web.application(url_params, self.resp_heads)
# 2.拼接数据然后发送给浏览器
# 构造响应头信息
# 空行
# 响应体
resp_head = 'HTTP/1.1 %s\r\n'%self.status_code
for field in self.resp_fields:
# 设置Content-Length的长度
resp_head += '%s:%s\r\n'%(field[0], field[1] if field[0] != 'Content-Length' else len(body.encode('utf-8')))
content = resp_head + '\r\n' + body
client_dict[client_fd].send(content.encode('utf-8'))
else:
# 返回静态数据
try:
f = open('./static%s'%filename, 'rb')
except Exception as e:
# 读取失败
body = 'Sorry! File not found!'
resp_head = 'HTTP/1.1 404 Not Found\r\n'
resp_head += 'Content-Length:%d\r\n'% len(body)
resp_head += '\r\n'
content = resp_head + body
client_dict[client_fd].send(content.encode('utf-8'))
# print('读取文件失败！')
logging.warning(e)
else:
# 读取成功
body = f.read()
resp_head = 'HTTP/1.1 200 OK\r\n'
resp_head += 'Content-Length:%d\r\n' % len(body)
resp_head += '\r\n'
try:
client_dict[client_fd].send(resp_head.encode('utf-8'))
client_dict[client_fd].send(body)
except Exception as e:
logging.warning(e)
def resp_heads(self, status_code, fields):
"""在框架中使用的函数，框架用来返回响应头信息"""
self.status_code = status_code
self.resp_fields = fields
def main():
"""程序入口"""
# 1.初始化服务器
server = WebServer()
# 2.运行服务器
server.run()
if __name__ == '__main__':
main()
框架部分：
# 服务器给数据,返回数据给服务器
import re
from urllib.request import unquote #解码中文,只用在浏览器自动对中文编码
import DBHelper
# 定义空字典,用来存储路径跟对应的函数引用
url_dict = dict()
# start_response框架给服务器传响应头的数据
# environ获取服务器传过来的文件路径
def application(environ, start_response):
"""返回具体展示的界面给服务器"""
start_response('200 OK', [('Content-Type', 'text/html;charset=utf-8'),
('Content-Length', '')]) # 返回响应头
# 根据不同的地址进行判断
file_name = environ['filename']
for key, func in url_dict.items():
match = re.match(key, file_name) # 地址跟规则一致
if match:
# 匹配到了
return func(match) # 调用匹配到的函数引用,返回匹配的页面内容
else:
# 说明没找到
return"不好意思，页面走丢了!"
# 装饰器传参,用来完成路由的功能
def route(url_address): # url_address表示页面的路径
"""目的自动添加路径跟匹配的函数到url字典中"""
def set_fun(func):
def call_fun(*args, **kwargs):
return func(*args, **kwargs)
# 根据不同的函数名称去添加到字典中
url_dict[url_address] = call_fun
return call_fun
return set_fun