正文

从零开始用Python实现HTTP服务器 手把手教你处理GET和POST请求的完整网络编程实例

/0 浏览量
0903

从零开始用Python实现HTTP服务器 手把手教你处理GET和POST请求的完整网络编程实例

嘿，朋友，坐下来聊聊。

你可能听说过HTTP服务器，也用过Nginx、Apache，或者像Flask、Django这样的框架。但你有没有想过，这些庞然大物到底是怎么工作的？今天咱们不绕弯子，就用最纯粹的Python，从零手搓一个能处理GET和POST请求的HTTP服务器。

我写这篇文章的时候，窗外下着小雨，我刚给自己泡了杯咖啡。说实话，第一次读懂HTTP协议的底层原理时，那种”原来如此”的快感，堪比小时候拆开收音机发现里面只有几块电路板。今天，咱们也来享受这种快乐。

先搞清楚HTTP长什么样

别急着写代码，先看看浏览器和服务器之间到底在传递什么。

打开浏览器，按F12，切到Network标签，随便访问一个网站，然后点击任意一个请求。你会看到类似这样的东西：

GET /index.html HTTP/1.1
Host: www.example.com
User-Agent: Mozilla/5.0 ...
Accept: text/html,application/xhtml+xml ...

这就是一个完整的HTTP请求。你看，它其实就是纯文本。没有花里胡哨的二进制，没有加密的外衣，就是几行文字，告诉服务器你想要什么。

  • 第一行是请求行：方法（GET）、路径（/index.html）、协议版本（HTTP/1.1）
  • 中间是请求头：一堆键值对，告诉服务器各种信息
  • 空行之后是请求体：GET请求一般没有，POST请求会有数据

服务器收到后，也会用同样的方式回复：

HTTP/1.1 200 OK
Content-Type: text/html
Content-Length: 1234

<html><body>Hello World</body></html>

第一行是状态行：协议版本、状态码（200表示成功）、状态描述。然后是响应头，空行，最后是响应体。

搞懂了这些，你就已经比80%的人更理解HTTP了。

Python的socket，就是我们的画布

HTTP的本质，就是网络连接。Python里处理网络连接的原始工具是socket模块。

先写一个最简单的服务器，感受一下：

import socket

# 创建TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 允许端口复用，避免重启时"Address already in use"
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 绑定到本地8080端口
server_socket.bind(('0.0.0.0', 8080))

# 开始监听， backlog=5 表示最多等待5个待处理的连接
server_socket.listen(5)

print("🎉 服务器已启动，访问 http://localhost:8080 试试")

while True:
    # accept() 会阻塞，直到有客户端连接
    client_socket, client_address = server_socket.accept()
    print(f"📥 收到连接：{client_address}")
    
    # 接收客户端发来的数据，最多1024字节
    request = client_socket.recv(1024).decode('utf-8')
    print("📨 收到请求：")
    print(request)
    
    # 简单回复
    response = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n<h1>Hello from my server!</h1>"
    client_socket.sendall(response.encode('utf-8'))
    
    # 关闭连接
    client_socket.close()

运行它，然后在浏览器访问http://localhost:8080，你会看到”Hello from my server!“。

但这段代码太粗糙了。它不解析请求，不管你是GET还是POST，它一律返回同样的内容。咱们来认真一点。

写一个真正的HTTP请求解析器

HTTP请求就是文本，咱们用字符串处理来解析它。核心思路：

  1. 拿到原始请求文本
  2. 按\r\n分割成行
  3. 第一行是请求行，拆出方法、路径、协议版本
  4. 后续行是请求头，拆成键值对
  5. 遇到空行后，剩余的是请求体（POST请求的数据）
def parse_http_request(raw_data: bytes) -> dict:
    """
    解析原始HTTP请求数据，返回结构化的请求信息
    """
    # 把原始字节解码成字符串
    request_text = raw_data.decode('utf-8', errors='replace')
    lines = request_text.split('\r\n')
    
    if not lines:
        return None
    
    # 解析请求行：METHOD PATH HTTP/1.1
    request_line = lines[0].split(' ')
    if len(request_line) < 2:
        return None
    
    method = request_line[0].upper()       # GET, POST, PUT, DELETE...
    raw_path = request_line[1]             # /api/users?id=1
    protocol = request_line[2] if len(request_line) > 2 else 'HTTP/1.1'
    
    # 解析路径和查询参数
    if '?' in raw_path:
        path, query_string = raw_path.split('?', 1)
    else:
        path = raw_path
        query_string = ''
    
    # 解析查询参数为字典
    query_params = {}
    if query_string:
        for pair in query_string.split('&'):
            if '=' in pair:
                key, value = pair.split('=', 1)
                # URL解码，把 %20 还原成空格，等等
                from urllib.parse import unquote
                query_params[unquote(key)] = unquote(value)
    
    # 解析请求头
    headers = {}
    body_start_index = 0
    for i, line in enumerate(lines[1:], start=1):
        if line == '':
            body_start_index = i + 1
            break
        if ':' in line:
            key, value = line.split(':', 1)
            headers[key.strip()] = value.strip()
    
    # 请求体（POST数据等）
    body = '\r\n'.join(lines[body_start_index:]) if body_start_index < len(lines) else ''
    
    return {
        'method': method,
        'path': path,
        'protocol': protocol,
        'query_params': query_params,
        'headers': headers,
        'body': body
    }

来，验证一下这个解析器好不好用。写个测试：

# 模拟一个GET请求
get_request = (
    b"GET /api/users?name=Alice&age=25 HTTP/1.1\r\n"
    b"Host: localhost:8080\r\n"
    b"User-Agent: TestClient/1.0\r\n"
    b"Accept: */*\r\n"
    b"\r\n"
)

# 模拟一个POST请求
post_request = (
    b"POST /api/login HTTP/1.1\r\n"
    b"Host: localhost:8080\r\n"
    b"Content-Type: application/x-www-form-urlencoded\r\n"
    b"Content-Length: 27\r\n"
    b"\r\n"
    b"username=admin&password=123456"
)

print(parse_http_request(get_request))
print(parse_http_request(post_request))

输出大概是：

{'method': 'GET', 'path': '/api/users', 'protocol': 'HTTP/1.1', 
 'query_params': {'name': 'Alice', 'age': '25'}, 
 'headers': {'Host': 'localhost:8080', 'User-Agent': 'TestClient/1.0', 'Accept': '*/*'}, 
 'body': ''}

{'method': 'POST', 'path': '/api/login', 'protocol': 'HTTP/1.1', 
 'query_params': {}, 
 'headers': {'Host': 'localhost:8080', 'Content-Type': 'application/x-www-form-urlencoded', 'Content-Length': '27'}, 
 'body': 'username=admin&password=123456'}

完美。解析器就位了。

构建响应生成器

解析了请求，接下来要生成响应。响应格式刚才已经说过了，咱们封装成一个函数：

def build_response(status_code: int, body: str, content_type: str = 'text/plain', extra_headers: dict = None) -> bytes:
    """
    构建HTTP响应
    """
    # 状态码映射
    status_messages = {
        200: 'OK',
        201: 'Created',
        204: 'No Content',
        301: 'Moved Permanently',
        302: 'Found',
        400: 'Bad Request',
        404: 'Not Found',
        405: 'Method Not Allowed',
        500: 'Internal Server Error'
    }
    
    status_msg = status_messages.get(status_code, 'Unknown')
    
    # 构建响应头
    headers = f"HTTP/1.1 {status_code} {status_msg}\r\n"
    headers += f"Content-Type: {content_type}\r\n"
    headers += f"Content-Length: {len(body.encode('utf-8'))}\r\n"
    headers += "Connection: close\r\n"  # 简单起见，每个请求结束后关闭连接
    
    if extra_headers:
        for key, value in extra_headers.items():
            headers += f"{key}: {value}\r\n"
    
    headers += "\r\n"  # 空行分隔头和体
    
    # 拼接响应
    response = headers + body
    return response.encode('utf-8')

路由系统 —— 这是服务器的灵魂

一个没有路由的服务器，就像一家没有菜单的餐厅 —— 不管你点啥，都给你上同样的菜。

咱们的路由系统要支持：

  • 按路径匹配（/api/users）
  • 按HTTP方法区分（GET /api/users 和 POST /api/users 做不同的事）
  • 简单易用

用字典来存路由，键是(method, path)，值是处理函数：

# 路由表：(method, path) -> handler_function
routes = {}

def route(method: str, path: str):
    """
    装饰器：注册路由
    用法：
        @route('GET', '/api/users')
        def get_users(request):
            return build_response(200, '{"users": [...]}')
    """
    def decorator(func):
        routes[(method.upper(), path)] = func
        return func
    return decorator

来点实际的 —— 实现几个API

现在咱们来写真正有用的路由。假设你在做一个简单的用户管理系统：

import json

# 模拟数据库
fake_database = {
    'users': [
        {'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'},
        {'id': 2, 'name': '李四', 'email': 'lisi@example.com'},
        {'id': 3, 'name': '王五', 'email': 'wangwu@example.com'}
    ]
}

@route('GET', '/')
def home(request):
    html = """
    <!DOCTYPE html>
    <html>
    <head><title>我的HTTP服务器</title></head>
    <body>
        <h1>🎉 欢迎来到我的HTTP服务器</h1>
        <ul>
            <li><a href="/api/users">查看用户列表 (GET /api/users)</a></li>
            <li><a href="/api/users/1">查看单个用户 (GET /api/users/1)</a></li>
            <li><a href="/api/users?name=张三">搜索用户 (GET /api/users?name=张三)</a></li>
        </ul>
        <h2>测试POST请求</h2>
        <form method="POST" action="/api/users">
            姓名：<input type="text" name="name"><br>
            邮箱：<input type="email" name="email"><br>
            <button type="submit">提交</button>
        </form>
    </body>
    </html>
    """
    return build_response(200, html, content_type='text/html')

@route('GET', '/api/users')
def get_users(request):
    """获取用户列表，支持 ?name=xxx 过滤"""
    name_filter = request['query_params'].get('name', '')
    
    users = fake_database['users']
    if name_filter:
        users = [u for u in users if name_filter in u['name']]
    
    response_body = json.dumps({'users': users}, ensure_ascii=False, indent=2)
    return build_response(200, response_body, content_type='application/json')

@route('GET', '/api/users/<int:user_id>')
def get_user(request, user_id: int):
    """获取单个用户"""
    user = next((u for u in fake_database['users'] if u['id'] == user_id), None)
    
    if user:
        return build_response(200, json.dumps(user, ensure_ascii=False), content_type='application/json')
    else:
        return build_response(404, json.dumps({'error': '用户不存在'}), content_type='application/json')

@route('POST', '/api/users')
def create_user(request):
    """创建新用户 —— 处理POST请求"""
    # 解析POST表单数据
    body = request['body']
    params = {}
    for pair in body.split('&'):
        if '=' in pair:
            key, value = pair.split('=', 1)
            from urllib.parse import unquote
            params[unquote(key)] = unquote(value)
    
    # 简单验证
    if not params.get('name') or not params.get('email'):
        return build_response(400, json.dumps({'error': '姓名和邮箱不能为空'}), content_type='application/json')
    
    # 创建用户
    new_id = max(u['id'] for u in fake_database['users']) + 1
    new_user = {
        'id': new_id,
        'name': params['name'],
        'email': params['email']
    }
    fake_database['users'].append(new_user)
    
    return build_response(201, json.dumps(new_user, ensure_ascii=False), content_type='application/json')

等等，你可能会问：<int:user_id> 这种路径参数是怎么匹配的？

好问题。咱们的路由匹配逻辑需要比简单的字典查找更聪明一点：

import re

def match_route(method: str, path: str) -> tuple:
    """
    匹配请求到对应的路由处理函数
    返回: (handler_function, matched_params_dict) 或 (None, None)
    """
    for (route_method, route_path), handler in routes.items():
        # 如果路由路径包含参数，如 /api/users/<int:user_id>
        if '<' in route_path:
            # 把路由路径转换成正则表达式
            regex_path = re.sub(r'<int:(\w+)>', r'(?P<\1>\d+)', route_path)
            regex_path = re.sub(r'<str:(\w+)>', r'(?P<\1>[^/]+)', regex_path)
            regex_path = re.sub(r'<(\w+)>', r'(?P<\1>[^/]+)', regex_path)
            
            match = re.fullmatch(regex_path, path)
            if match and method == route_method:
                return handler, match.groupdict()
        else:
            # 普通路径匹配
            if path == route_path and method == route_method:
                return handler, {}
    
    return None, None

现在路由匹配就支持动态路径了。

组装主循环

前面所有的零件都准备好了，现在把它们组装起来：

import socket
import re
import json
from urllib.parse import unquote

# ===== 前面所有的代码（parse_http_request, build_response, route装饰器, 路由函数, match_route）=====

def handle_client(client_socket: socket.socket):
    """处理单个客户端连接"""
    try:
        # 接收请求数据，设置超时防止无限阻塞
        client_socket.settimeout(5.0)
        raw_data = b''
        while True:
            chunk = client_socket.recv(4096)
            if not chunk:
                break
            raw_data += chunk
            # 如果收到了完整的请求（通过Content-Length判断是否有body）
            if b'\r\n\r\n' in raw_data:
                # 提取请求头，看看Content-Length是多少
                header_part = raw_data.split(b'\r\n\r\n')[0].decode('utf-8', errors='replace')
                content_length = 0
                for line in header_part.split('\r\n'):
                    if line.lower().startswith('content-length:'):
                        content_length = int(line.split(':')[1].strip())
                        break
                body_part = raw_data.split(b'\r\n\r\n')[1] if b'\r\n\r\n' in raw_data else b''
                if len(body_part) >= content_length:
                    break
    except socket.timeout:
        pass
    except Exception:
        pass
    
    if not raw_data:
        client_socket.close()
        return
    
    # 解析请求
    request = parse_http_request(raw_data)
    if not request:
        response = build_response(400, '<h1>400 Bad Request</h1>')
        client_socket.sendall(response)
        client_socket.close()
        return
    
    print(f"\n{'='*50}")
    print(f"📨 {request['method']} {request['path']}")
    print(f"   来自: {client_socket.getpeername()}")
    if request['query_params']:
        print(f"   参数: {request['query_params']}")
    if request['body']:
        print(f"   请求体: {request['body']}")
    print(f"{'='*50}")
    
    # 路由匹配
    handler, params = match_route(request['method'], request['path'])
    
    if handler:
        try:
            # 把路径参数也传入请求字典
            request['path_params'] = params or {}
            # 调用路由处理函数
            if params:
                response = handler(request, **params)
            else:
                response = handler(request)
        except Exception as e:
            print(f"❌ 处理请求时出错: {e}")
            response = build_response(500, f'<h1>500 Internal Server Error</h1><p>{e}</p>')
    else:
        # 没匹配到路由，检查方法是否匹配但路径不匹配
        method_matched = any(m == request['method'] for m, p in routes.keys())
        if method_matched:
            response = build_response(404, '<h1>404 Not Found</h1>')
        else:
            response = build_response(405, f'<h1>405 Method Not Allowed</h1><p>{request["method"]} not supported for this path</p>')
    
    # 发送响应
    client_socket.sendall(response)
    client_socket.close()


def main():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定端口
    host = '0.0.0.0'
    port = 8080
    server_socket.bind((host, port))
    server_socket.listen(100)  # 支持100个等待连接
    
    print(f"🚀 HTTP服务器已启动")
    print(f"📍 地址: http://{host}:{port}")
    print(f"📋 已注册的路由:")
    for (method, path), handler in routes.items():
        print(f"   {method} {path} -> {handler.__name__}")
    print(f"\n按 Ctrl+C 停止服务器\n")
    
    try:
        while True:
            client_socket, client_address = server_socket.accept()
            print(f"🔗 新连接: {client_address}")
            handle_client(client_socket)
    except KeyboardInterrupt:
        print("\n👋 服务器已停止")
    finally:
        server_socket.close()

if __name__ == '__main__':
    main()

跑起来，试试看

保存代码为 http_server.py，运行：

python http_server.py

输出应该是：

🚀 HTTP服务器已启动
📍 地址: http://0.0.0.0:8080
📋 已注册的路由:
   GET / -> home
   GET /api/users -> get_users
   GET /api/users/<int:user_id> -> get_user
   POST /api/users -> create_user

🔗 新连接: ('127.0.0.1', 54321)
==================================================
📨 GET /
   来自: ('127.0.0.1', 54321)
==================================================

然后：

  • 浏览器访问 http://localhost:8080，看到欢迎页面
  • 访问 http://localhost:8080/api/users，看到用户列表JSON
  • 访问 http://localhost:8080/api/users/1，看到单个用户
  • 在页面上的表单填写姓名和邮箱，提交，你会看到POST请求被正确处理，新用户被添加到”数据库”

用curl测试POST：

curl -X POST http://localhost:8080/api/users \
  -d "name=赵六&email=zhaoliu@example.com"

返回：

{"id": 4, "name": "赵六", "email": "zhaoliu@example.com"}

一些你可能想知道的细节

为什么用SO_REUSEADDR？

不设置这个的话，服务器关闭后立刻重启，可能会遇到”Address already in use”错误。这是因为TCP连接关闭后，端口不会立即释放，还需要等待一段时间（TIME_WAIT状态）。SO_REUSEADDR允许我们立即重用端口。

为什么响应头里写Connection: close？

最简单的做法。每个请求建立一个连接，处理完就关闭。真实的生产服务器会用Keep-Alive让一个连接处理多个请求，但这需要更复杂的逻辑。先理解核心，再考虑优化。

GET和POST在这里有什么区别？

在代码层面，区别就是：GET请求的数据在URL查询参数里（request['query_params']），POST请求的数据在请求体里（request['body']）。服务器对两者的处理方式不同，但最终都是通过路由匹配到不同的处理函数。

这个服务器能扛住多少人用？

说实话，几乎不能。它是单线程的，一次只能处理一个请求。第二个请求必须等第一个处理完。但咱们的目标不是造一个生产级服务器，而是理解HTTP的本质。真要上生产，用Gevent、Asyncio，或者直接上Nginx + uWSGI。

最后说几句

写这个服务器的时候，我最大的感受是：HTTP其实没那么神秘。

浏览器发一段文本，服务器回一段文本。就这么简单。所有那些复杂的框架、中间件、路由系统，底层都是这个机制。理解了它，再看Flask、FastAPI的源码，就会觉得”不过如此”。

你完全可以在这个基础上继续扩展：加WebSocket支持、做静态文件服务、实现会话管理、加上HTTPS……每一步都是在现有代码上加东西，而不是推倒重来。

网络编程的世界，的大门已经对你敞开了。祝你玩得开心。

-- 展开阅读全文 --