目录
Python实例题
题目
要求：
解题思路：
代码实现：
Python实例题
题目
简单的 Web 服务器
要求：
- 使用 Python 的
socket模块实现一个简单的 HTTP 服务器。 - 支持以下功能：
- 处理 GET 和 POST 请求
- 静态文件服务（HTML、CSS、JavaScript、图片等）
- 简单的路由系统
- 错误处理（404、500 等）
- 添加多线程支持，以处理并发请求。
解题思路：
- 使用
socket建立网络连接。 - 解析 HTTP 请求头，处理不同类型的请求。
- 使用线程池处理并发连接。
代码实现：
import socket
import threading
import os
import mimetypes
from urllib.parse import urlparse, parse_qsclass WebServer:def __init__(self, host='localhost', port=8080, document_root='www'):self.host = hostself.port = portself.document_root = document_rootself.routes = {'GET': {},'POST': {}}self.server_socket = Noneself.running = Falsedef route(self, method, path, handler):"""注册路由处理函数"""self.routes[method][path] = handlerdef start(self):"""启动服务器"""self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server_socket.bind((self.host, self.port))self.server_socket.listen(5)self.running = Trueprint(f"服务器运行中: http://{self.host}:{self.port}")try:while self.running:client_socket, client_address = self.server_socket.accept()print(f"接受来自 {client_address} 的连接")# 创建线程处理请求client_handler = threading.Thread(target=self.handle_client,args=(client_socket,))client_handler.start()except KeyboardInterrupt:self.stop()def stop(self):"""停止服务器"""self.running = Falseif self.server_socket:self.server_socket.close()print("服务器已停止")def handle_client(self, client_socket):"""处理客户端请求"""try:# 接收请求数据request_data = client_socket.recv(1024).decode('utf-8')if not request_data:return# 解析请求request_lines = request_data.splitlines()request_line = request_lines[0]method, path, _ = request_line.split(' ', 2)# 解析查询参数parsed_url = urlparse(path)path = parsed_url.pathquery_params = parse_qs(parsed_url.query)# 处理 POST 请求体post_data = {}if method == 'POST':# 找到请求头和请求体的分隔线body_index = request_data.find('\r\n\r\n')if body_index != -1:body = request_data[body_index + 4:]post_data = parse_qs(body)# 处理路由if path in self.routes[method]:response = self.routes[method][path](query_params, post_data)else:# 静态文件服务response = self.serve_static_file(path)# 发送响应client_socket.sendall(response)except Exception as e:print(f"处理请求时出错: {e}")error_response = self.generate_error_response(500)client_socket.sendall(error_response)finally:# 关闭连接client_socket.close()def serve_static_file(self, path):"""提供静态文件服务"""# 处理根路径if path == '/':path = '/index.html'# 构建文件路径file_path = os.path.join(self.document_root, path[1:])# 检查文件是否存在if not os.path.exists(file_path):return self.generate_error_response(404)# 检查是否是目录if os.path.isdir(file_path):return self.generate_error_response(403)try:# 读取文件内容with open(file_path, 'rb') as file:content = file.read()# 确定 MIME 类型content_type, _ = mimetypes.guess_type(file_path)if not content_type:content_type = 'application/octet-stream'# 生成响应头response_headers = (f'HTTP/1.1 200 OK\r\n'f'Content-Type: {content_type}\r\n'f'Content-Length: {len(content)}\r\n'f'\r\n').encode('utf-8')return response_headers + contentexcept Exception as e:print(f"读取文件时出错: {e}")return self.generate_error_response(500)def generate_error_response(self, status_code):"""生成错误响应"""status_messages = {404: 'Not Found',403: 'Forbidden',500: 'Internal Server Error'}status_message = status_messages.get(status_code, 'Error')# 生成错误页面error_page = (f'<html><body><h1>{status_code} {status_message}</h1></body></html>').encode('utf-8')# 生成响应头response_headers = (f'HTTP/1.1 {status_code} {status_message}\r\n'f'Content-Type: text/html\r\n'f'Content-Length: {len(error_page)}\r\n'f'\r\n').encode('utf-8')return response_headers + error_pagedef main():# 创建服务器实例server = WebServer(document_root='www')# 注册动态路由def home_handler(query_params, post_data):response_body = '<html><body><h1>欢迎来到我的网站!</h1></body></html>'.encode('utf-8')response_headers = ('HTTP/1.1 200 OK\r\n''Content-Type: text/html\r\n'f'Content-Length: {len(response_body)}\r\n''\r\n').encode('utf-8')return response_headers + response_bodydef echo_handler(query_params, post_data):message = query_params.get('message', [''])[0]response_body = f'<html><body><p>你发送的消息是: {message}</p></body></html>'.encode('utf-8')response_headers = ('HTTP/1.1 200 OK\r\n''Content-Type: text/html\r\n'f'Content-Length: {len(response_body)}\r\n''\r\n').encode('utf-8')return response_headers + response_bodyserver.route('GET', '/', home_handler)server.route('GET', '/echo', echo_handler)# 启动服务器try:server.start()except KeyboardInterrupt:passif __name__ == "__main__":main()
