第4章：HTTP 与 Web 服务器

元数据卡

字段	值
难度	(进阶级)
前置依赖	Vol 4 第 3 章（TCP）、基本 socket 编程
关键词	HTTP/1.1、请求/响应格式、方法、状态码、Cache-Control、Cookie、CORS、持久连接、管线化、迷你 Web 服务器
核心技能	读懂任何一条 HTTP 报文；用 Python 完整实现一个 HTTP/1.1 服务器；配置缓存策略和 CORS 头

你在哪

"TCP 帮你铺好了可靠传送咒的路，现在你可以在驿道上跑法术信函了。HTTP 就是驿道世界的通用信件格式——法师的观测镜用 HTTP 请求法术卷轴，驿道信标塔用 HTTP 返回内容。你写的第一个信标塔服务器就从这里开始。"

第 3 章你钻进了 TCP 的骨髓——首部、法力握手、卷轴窗口、拥塞控制。HTTP 就是站在 TCP 肩膀上长出来的"应用层咒语之王"。

你的法师观测镜每天发出成百上千个 HTTP 法术请求。但你知道 GET 和 POST 的咒语差异到底在哪里吗？304 Not Modified 是怎么省掉传送的？Cache-Control: no-cache 到底行不行？

本章不是"用 requests 库调用法术信标"——那是日常。本章是：你写一个 socket 法术接口，亲手拼出 HTTP 法术报文，然后看到信标塔回复原始魔力字节，你再一行行解析回去。

你的任务

本章分层
必读：手写并读懂 HTTP 请求/响应报文（原始格式），理解方法（GET/POST/PUT/DELETE）语义、状态码分类、核心 Headers（Content-Length、Cache-Control、Cookie、CORS）
选读：持久连接 vs 短连接、管线化（历史知识）
🛠 项目实战：迷你 Web 服务器（完整实现，位于本章末尾）
本章不会要求你掌握
HTTP/2 的 HPACK 头压缩细节
WebSocket 升级握手
OAuth 2.0 / OpenID Connect 流程

写一个 HTTP/1.1 请求和响应的 raw 报文，能逐行解释每一行
用 Python 实现一个完整的迷你 Web 服务器（支持 GET/POST/PUT/DELETE）——标记为项目实战
理解 HTTP 方法的安全性和幂等性语义
读懂每类状态码（1xx-5xx）的意图
掌握核心 Headers：Cache-Control、Cookie、CORS
了解 HTTP 持久连接和管线化的概念（管线化作为历史背景了解即可）

遭遇战 → 获得技能

4.1 HTTP 请求：你每天发的"信"长什么样

打开 DevTools → Network → 复制一个请求为 cURL。它真正的原始格式可能让你吃惊——因为 DevTools 替你把头给藏了。

GET /index.html HTTP/1.1
Host: www.example.com
User-Agent: Mozilla/5.0 ...
Accept: text/html,application/xhtml+xml,...
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Connection: keep-alive

对，HTTP 是纯文本的。

💬 给 Java/C++ 工程师 Java 的 HttpURLConnection 和 Spring 的 RestTemplate 把 HTTP 封装得像 RPC——你写 restTemplate.getForObject(url, String.class) 就完事了。C++ 的 libcurl 则更贴近原始报文：curl_easy_setopt(curl, CURLOPT_URL, url) + curl_easy_perform(curl)。但你写过一个能自己拼 HTTP 报文的服务器吗？在 Java 里做这件事需要 ServerSocket + InputStreamReader —— 和 Python 没有本质区别，只是多了一层 checked exception 的包裹。在 C++ 里你能更清晰地看到 send() 和 recv() 循环。

请求行

GET /path?query=value HTTP/1.1
↑方法  ↑请求目标（URI+查询）     ↑HTTP 版本

Headers（键值对，每行一个）

Header-Name: value

大小写不敏感——Content-Type 和 content-type 等价。但习惯用 PascalCase。

空行 + Body

POST /api/data HTTP/1.1
Host: example.com
Content-Type: application/json
Content-Length: 27

{"key": "value", "id": 42}

Content-Length 告诉接收方 Body 有多长——TCP 是流，没有"分界标记"。Content-Length 就是分界符。

Python 解析 raw HTTP 请求：

python

import re
from typing import Dict, Optional

def parse_http_request(raw: bytes) -> dict:
    """把从 socket recv 收到的 bytes 解析为结构化 dict"""
    # 找到首部和 Body 的分界线（\r\n\r\n）
    header_end = raw.find(b'\r\n\r\n')
    if header_end == -1:
        raise ValueError("不完整的 HTTP 请求——缺少空行分隔")

    header_part = raw[:header_end].decode('utf-8', errors='replace')
    body_part = raw[header_end + 4:]

    lines = header_part.split('\r\n')
    request_line = lines[0]
    method, path, version = request_line.split(' ', 2)

    headers = {}
    for line in lines[1:]:
        if ':' in line:
            key, val = line.split(':', 1)
            headers[key.strip().lower()] = val.strip()

    return {
        'method': method.upper(),
        'path': path,
        'version': version.strip(),
        'headers': headers,
        'body': body_part,
    }

# 测试
raw_req = (
    b"POST /submit HTTP/1.1\r\n"
    b"Host: localhost:8080\r\n"
    b"Content-Type: application/json\r\n"
    b"Content-Length: 27\r\n"
    b"\r\n"
    b'{"user": "steven", "score": 100}'
)
parsed = parse_http_request(raw_req)
print(f"方法: {parsed['method']}, 路径: {parsed['path']}")
print(f"Body: {parsed['body']}")

4.2 HTTP 响应：服务器回的信

HTTP/1.1 200 OK
Date: Mon, 23 Jun 2026 16:00:00 GMT
Content-Type: text/html; charset=utf-8
Content-Length: 1234
Cache-Control: max-age=3600
Last-Modified: Mon, 22 Jun 2026 12:00:00 GMT

<!DOCTYPE html>
<html><body><h1>Hello</h1></body></html>

状态行解析：

HTTP/1.1  200      OK
↑版本      ↑状态码  ↑原因短语

4.3 HTTP 方法：动词的语义

HTTP 方法的语义不是"你用哪个就哪个"。它承载了安全和幂等的契约。

方法	可缓存	Body	典型用途
GET		不应有	获取资源
HEAD		不应有	只查元数据
OPTIONS		不应有	查 CORS/服务器能力
POST	看响应头	有	创建资源（提交表单）
PUT		有	替换资源（全量更新）
PATCH		有	部分更新资源
DELETE		可变	删除资源

安全：不改变服务器状态。你可以爬取十个 GET，服务器数据不受影响。
幂等：执行一次和一百次的效果相同。DELETE 幂等——删完再删，服务器返回 404 而不是创建什么。POST 不幂等——提交订单一次成功，再提交一次就可能重复下单。

方法差异的底层逻辑

为什么需要 PUT vs POST 的区分？

POST /api/users     → 每次创建不同的用户（不幂等，URL 由服务器分配）
PUT /api/users/42   → 覆盖 /api/users/42 的内容（幂等，URL 由客户端指定）

RESTful API 设计遵循这个原则。违反它（比如用 POST 做更新）不算错，但会让客户端难以实现重试逻辑和缓存。

4.4 状态码：服务器的"脸色"

五大家族

分类	范围	含义	经典代表
1xx	100-199	信息性	100 Continue（继续发送 Body）
2xx	200-299	成功	200 OK, 201 Created, 204 No Content
3xx	300-399	重定向	301 Moved Permanently, 302 Found, 304 Not Modified
4xx	400-499	客户端错误	400 Bad Request, 401 Unauthorized, 403 Forbidden, 404 Not Found, 429 Too Many Requests
5xx	500-599	服务器错误	500 Internal Server Error, 502 Bad Gateway, 503 Service Unavailable

关键状态码深度理解

304 Not Modified：条件请求。

客户端发: GET /logo.png
          If-Modified-Since: Mon, 22 Jun 2026 12:00:00 GMT

服务器对比后，文件没变 → 304（0 bytes body）
客户端知道直接使用缓存。

如果变了 → 200 + 新文件。

304 省的不是请求次数（请求一样发出了），省的是响应体的网络传输。

429 Too Many Requests：限流。

HTTP/1.1 429 Too Many Requests
Retry-After: 60    ← 等 60 秒再试

API 客户端需要尊重 Retry-After，不然拿不到数据还白白消耗资源。

502 Bad Gateway vs 503 Service Unavailable：

502：上游（比如你的应用服务器）返回了异常响应
503：服务器太忙或宕机中，自己知道自己有问题

状态码选择速查

python

from enum import IntEnum

class HttpStatus(IntEnum):
    OK = 200
    CREATED = 201
    ACCEPTED = 202
    NO_CONTENT = 204
    MOVED_PERMANENTLY = 301
    NOT_MODIFIED = 304
    BAD_REQUEST = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    METHOD_NOT_ALLOWED = 405
    CONFLICT = 409
    UNPROCESSABLE_ENTITY = 422
    TOO_MANY_REQUESTS = 429
    INTERNAL_ERROR = 500
    BAD_GATEWAY = 502
    SERVICE_UNAVAILABLE = 503

    @property
    def reason(self) -> str:
        mapping = {
            200: "OK",
            201: "Created",
            400: "Bad Request",
            401: "Unauthorized",
            404: "Not Found",
            500: "Internal Server Error",
            # ... 其他同理
        }
        # 标准原因短语可从 http.client 获取
        import http.client
        return http.client.responses.get(self.value, "Unknown")

4.5 Headers：HTTP 的"元力量"

Cache-Control：如何对客户端和代理说"缓存多久"

Cache-Control 是 HTTP/1.1 定义的缓存指令，取代了 HTTP/1.0 的 Expires 和 Pragma。

指令	含义	示例
no-cache	验证后才能用缓存。每次请求都要问服务器"缓存过期没"	`Cache-Control: no-cache`
no-store	完全不要缓存（敏感数据）	`Cache-Control: no-store`
max-age=<秒>	从响应生成起，缓存有效多久	`Cache-Control: max-age=3600`（1 小时）
public	任何缓存（包括 CDN/代理）都可以存	`Cache-Control: public, max-age=86400`
private	只有最终用户缓存可存（CDN 不能存）	`Cache-Control: private, max-age=3600`
must-revalidate	过期后必须验证，不能直接用过期缓存	`Cache-Control: must-revalidate`

缓存策略选择术：

python

def cache_control_for_resource(resource_type: str) -> str:
    """根据资源类型返回推荐的 Cache-Control 指令"""
    config = {
        'static_asset': 'public, max-age=31536000, immutable',   # 1年，不变
        'api_personal': 'private, no-cache',                      # 用户相关
        'api_general': 'public, max-age=60, must-revalidate',     # 通用数据
        'html_page': 'no-cache',                                  # HTML 内容
        'sensitive': 'no-store',                                  # 不存
    }
    return config.get(resource_type, 'no-cache')

常见误解：no-cache 不是不缓存！ 它说的是：缓存必须去服务器验证有效性（用 ETag/If-Modified-Since），验证通过后才能使用。真正不缓存是 no-store。

HTTP 是无状态的。服务器无法区分两次请求是否来自同一个浏览器。Cookie 是在 HTTP 头上"附加的状态"。

服务器 → 客户端:
Set-Cookie: session_id=abc123; Path=/; HttpOnly; Secure; SameSite=Lax

客户端 → 服务器（后续请求自动带）:
Cookie: session_id=abc123

Cookie 属性	用途
`Expires` / `Max-Age`	Cookie 存活期
`Domain`	哪些域名发这个 Cookie
`Path`	哪些路径附带 Cookie
`Secure`	只在 HTTPS 下发送
`HttpOnly`	JavaScript 无法读取（防 XSS）
`SameSite=Lax\|Strict\|None`	CSRF 防护，控制跨站请求是否带 Cookie

SameSite 的进化史：

Chrome 80+ 默认 SameSite=Lax——第三方请求（例如从 A 站链接跳转到 B站）会带 Cookie，但 POST/iframe/script 发起的跨站请求不带
SameSite=Strict——任何跨站请求都不带 Cookie（最严，但也最烦——从邮件里的链接跳过去，Cookie 丢了，用户需重新登录）
SameSite=None; Secure——跨站带 Cookie（无防护，只在 HTTPS 下允许）

💬 给 Java 工程师javax.servlet.http.Cookie 和 Spring 的 ResponseCookie.from("name", "value").httpOnly(true).secure(true).sameSite("Lax").build()。本质都是生成相同的 Set-Cookie 头。Cookie 跨域问题是统一用 CSRF Token（Spring Security 的 CsrfFilter）来解决的——把 SameSite 和 CSRF Token 搭配使用，防御深度才够。

CORS：跨域不是"安全问题"，是浏览器帮你做的限制

CORS（Cross-Origin Resource Sharing）不是后端安全机制****——它是浏览器对 非同源请求 的默认阻止，不是服务器。

同源：协议、域名、端口三者都相同。

页面在 https://myapp.com
请求 https://api.other.com/data
→ 跨域！浏览器默认不让读响应

CORS 流程：

简单请求（GET/HEAD/POST 且 Content-Type 是 form-encoded）：浏览器直接发请求，服务器回复 Access-Control-Allow-Origin: https://myapp.com。如果没这个头，浏览器不让 JS 读取响应（请求已经发出去了！但 JS 拿不到结果）。
预检请求（Preflighted，PUT/DELETE/自定义 Content-Type）：浏览器先发一个 OPTIONS 请求问服务器"我可以跨域吗？"

OPTIONS /data HTTP/1.1
Origin: https://myapp.com
Access-Control-Request-Method: PUT
Access-Control-Request-Headers: X-Custom-Header

HTTP/1.1 204 No Content
Access-Control-Allow-Origin: https://myapp.com
Access-Control-Allow-Methods: PUT, DELETE
Access-Control-Allow-Headers: X-Custom-Header
Access-Control-Max-Age: 86400

预检通过后，浏览器才发真正的 PUT 请求。

python

import json

def cors_response(origin: str, allowed_origins: list) -> dict:
    """生成 CORS 响应头"""
    headers = {}
    if origin in allowed_origins:
        headers['Access-Control-Allow-Origin'] = origin
    # 否则用通配符（仅公开 API 且不含凭据时安全）
    elif '*' in allowed_origins:
        headers['Access-Control-Allow-Origin'] = '*'
    else:
        return headers  # 不加 CORS 头，浏览器会拦截

    headers.update({
        'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
        'Access-Control-Allow-Headers': 'Content-Type, Authorization, X-Requested-With',
        'Access-Control-Max-Age': '86400',
    })
    return headers

# 在响应中应用
def make_response(body: dict, status: int = 200, cors_origin: str = '*') -> bytes:
    headers = cors_response(cors_origin, ['*', 'http://localhost:3000'])
    headers['Content-Type'] = 'application/json'
    response = f"HTTP/1.1 {status} {_reason(status)}\r\n"
    for k, v in headers.items():
        response += f"{k}: {v}\r\n"
    response += "\r\n"
    response += json.dumps(body)
    return response.encode()

4.6 HTTP 持久连接与管线化：别为每个请求新建 TCP

管线化：历史背景 — 以下「管线化」内容作为历史知识了解即可。它在现实中从未大规模普及，HTTP/2 的多路复用已经用更好的方式解决了同样的问题。理解它的存在和为什么失败更有价值。

短连接时代（HTTP/1.0 默认）

每次 HTTP 请求都新建 TCP 连接。

[打开 TCP] → HTTP 请求 ① → 响应 → [关闭 TCP]
[打开 TCP] → HTTP 请求 ② → 响应 → [关闭 TCP]
[打开 TCP] → HTTP 请求 ③ → 响应 → [关闭 TCP]

一个页面 100 个资源 → 100 个 TCP 连接 → 三次握手开销巨大。

持久连接（HTTP/1.1 默认）

[打开 TCP]
  HTTP 请求 ① → 响应
  HTTP 请求 ② → 响应
  HTTP 请求 ③ → 响应
[保持 TCP 空闲一段时间后关闭]

Connection: keep-alive（HTTP/1.1 默认，无需特意写）。服务器通常 60 秒超时后关闭空闲连接。

减少了三次握手的开销，但响应还有一个队头阻塞问题。

管线化——HTTP/1.1 的尝试与失败

[打开 TCP]
  请求 ① → 请求 ② → 请求 ③  (不等响应就发下一个)
  响应 ① → 响应 ② → 响应 ③  (服务器按序回复)

问题：服务器必须按序回复。如果 ① 很慢（查询数据库 2 秒），②③ 即使处理好了也得等。

这被称为HTTP 队头阻塞（Head-of-Line Blocking）。

为什么管线化没普及：

大部分服务器和代理实现有 bug
HTTP/1.1 的解决方案是：打开多个 TCP 连接（通常 6 个），每个连接独立处理
HTTP/2 用多路复用彻底解决了它

Python 验证——单连接 vs 多连接：

python

import socket
import time
from typing import List

class HTTPBench:
    def __init__(self, host: str, port: int = 80):
        self.host = host
        self.port = port

    def single_connection(self, paths: List[str]) -> float:
        """一个 TCP 连接顺序请求多个资源"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.host, self.port))
        start = time.perf_counter()

        for path in paths:
            req = f"GET {path} HTTP/1.1\r\nHost: {self.host}\r\nConnection: keep-alive\r\n\r\n"
            sock.sendall(req.encode())
            resp = b""
            while True:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                resp += chunk
                # 简化：只收第一个响应（不完整但够验证时序）
                if b"\r\n\r\n" in resp and b"Content-Length:" not in resp:
                    # 没有 Content-Length 时停在第一个响应后
                    # 实际应该解析 Content-Length 知道何时读完
                    pass

        elapsed = time.perf_counter() - start
        sock.close()
        return elapsed

    def multi_connections(self, paths: List[str]) -> float:
        """为每个资源建立独立 TCP 连接"""
        import concurrent.futures

        def fetch(path: str) -> None:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            sock.connect((self.host, self.port))
            req = f"GET {path} HTTP/1.1\r\nHost: {self.host}\r\nConnection: close\r\n\r\n"
            sock.sendall(req.encode())
            sock.recv(4096)
            sock.close()

        start = time.perf_counter()
        with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
            pool.map(fetch, paths)
        return time.perf_counter() - start

# bench = HTTPBench("httpbin.org")
# t1 = bench.single_connection(["/get", "/ip", "/uuid"])
# t2 = bench.multi_connections(["/get", "/ip", "/uuid"])
# print(f"单连接: {t1:.3f}s, 多连接: {t2:.3f}s")

项目实战：写一个迷你 Web 服务器

是时候把所有知识串起来了。用纯 socket（不用 Flask/FastAPI）写一个支持 GET/POST/PUT/DELETE 的 HTTP/1.1 服务器。

🛠 本节是完整的项目实战，建议你亲自把代码敲一遍运行。理解了这节，你对 HTTP 的理解就再也不会停留在框架层面了。

python

#!/usr/bin/env python3
"""
mini_httpd.py — 迷你 HTTP/1.1 服务器
支持：GET/POST/PUT/DELETE、JSON 响应、CORS、持久连接、404/405/500
"""
import json
import socket
import os
import mimetypes
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple
from urllib.parse import unquote, parse_qs
from datetime import datetime
import signal

# ──── 路由注册 ────
RouteHandler = Callable[[dict, bytes], Tuple[int, dict]]

class MiniRouter:
    """路由注册与匹配"""

    def __init__(self):
        self.routes: Dict[str, Dict[str, RouteHandler]] = {}
        # routes['GET']['/path'] = handler
        # routes['POST']['/path'] = handler

    def add(self, method: str, path: str, handler: RouteHandler):
        self.routes.setdefault(method.upper(), {})[path] = handler

    def get(self, path: str):
        return lambda f: self.add('GET', path, f)

    def post(self, path: str):
        return lambda f: self.add('POST', path, f)

    def put(self, path: str):
        return lambda f: self.add('PUT', path, f)

    def delete(self, path: str):
        return lambda f: self.add('DELETE', path, f)

    def match(self, method: str, path: str) -> Optional[Tuple[str, RouteHandler]]:
        """精确匹配路由路径"""
        method_routes = self.routes.get(method.upper(), {})
        handler = method_routes.get(path)
        if handler:
            return (path, handler)
        # TODO: 可以扩展为模式匹配如 /api/users/:id
        return None


# ──── HTTP 协议操作 ────

def _reason(status: int) -> str:
    reasons = {
        200: "OK", 201: "Created", 204: "No Content",
        301: "Moved Permanently", 304: "Not Modified",
        400: "Bad Request", 401: "Unauthorized", 403: "Forbidden",
        404: "Not Found", 405: "Method Not Allowed",
        409: "Conflict", 422: "Unprocessable Entity", 429: "Too Many Requests",
        500: "Internal Server Error", 502: "Bad Gateway", 503: "Service Unavailable",
    }
    return reasons.get(status, "Unknown Status")


def parse_http_request(data: bytes) -> dict:
    """将收到的 raw bytes 解析为结构化 dict"""
    # ... (见 4.1 节，此处略去实现)
    header_end = data.find(b'\r\n\r\n')
    if header_end == -1:
        return {'method': 'GET', 'path': '/', 'version': 'HTTP/1.1', 'headers': {}, 'body': b''}

    header_part = data[:header_end].decode('utf-8', errors='replace')
    body_part = data[header_end + 4:]

    lines = header_part.split('\r\n')
    method, path, version = lines[0].split(' ', 2)

    headers = {}
    for line in lines[1:]:
        if ':' in line:
            key, val = line.split(':', 1)
            headers[key.strip().lower()] = val.strip()

    path_only = path.split('?')[0] if '?' in path else path
    return {
        'method': method.upper(),
        'raw_path': path,
        'path': unquote(path_only),
        'version': version,
        'headers': headers,
        'body': body_part,
    }


def build_response(status: int, body: bytes = b'', headers: dict = None) -> bytes:
    """构建 HTTP 响应 bytes"""
    hdrs = headers or {}
    hdrs.setdefault('Content-Type', 'text/plain; charset=utf-8')
    hdrs.setdefault('Date', datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT'))
    hdrs.setdefault('Connection', 'keep-alive')

    if body:
        hdrs['Content-Length'] = str(len(body))

    # 状态行
    resp = f"HTTP/1.1 {status} {_reason(status)}\r\n".encode()
    for k, v in hdrs.items():
        resp += f"{k}: {v}\r\n".encode()
    resp += b"\r\n"
    resp += body
    return resp


def json_response(status: int, data: dict, cors: bool = True) -> bytes:
    """构建 JSON 响应"""
    body = json.dumps(data, ensure_ascii=False).encode('utf-8')
    headers = {'Content-Type': 'application/json; charset=utf-8'}
    if cors:
        headers['Access-Control-Allow-Origin'] = '*'
    return build_response(status, body, headers)


# ──── 应用层逻辑 ────

class MiniApp:
    """迷你 Web 应用对象"""

    def __init__(self):
        self.router = MiniRouter()
        self.middleware: List[Callable] = []
        self._register_defaults()

    def _register_defaults(self):
        """注册一个默认的数据存储"""

        store = {
            'message': 'Welcome to MiniHTTPD!',
            'items': [],
        }

        @self.router.get('/')
        def root(req, body):
            return (200, store['items'] + [store['message']])

        @self.router.get('/api/items')
        def list_items(req, body):
            return (200, store['items'])

        @self.router.post('/api/items')
        def create_item(req, body):
            try:
                item = json.loads(body.decode('utf-8'))
            except (json.JSONDecodeError, UnicodeDecodeError):
                return (400, {"error": "无效的 JSON"})
            store['items'].append(item)
            return (201, {"message": "已创建", "item": item})

        @self.router.put('/api/items')
        def replace_items(req, body):
            try:
                store['items'] = json.loads(body.decode('utf-8'))
            except (json.JSONDecodeError, UnicodeDecodeError):
                return (400, {"error": "无效的 JSON"})
            return (200, {"message": "已替换", "items": store['items']})

        @self.router.delete('/api/items')
        def delete_items(req, body):
            store['items'].clear()
            return (204, {})

    def handle_request(self, raw: bytes) -> bytes:
        """处理一条 HTTP 请求，返回响应 bytes"""
        try:
            req = parse_http_request(raw)
        except Exception:
            return json_response(400, {"error": "无法解析请求"})

        method = req['method']
        path = req['path']

        # OPTIONS → CORS 预检处理
        if method == 'OPTIONS':
            headers = {
                'Allow': 'GET, POST, PUT, DELETE, OPTIONS',
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
                'Access-Control-Allow-Headers': 'Content-Type, Authorization',
                'Access-Control-Max-Age': '86400',
            }
            return build_response(204, b'', headers)

        # 路由匹配
        match = self.router.match(method, path)
        if not match:
            # 检查方法是否只对路径不存在
            if self.router.match('GET', path) or self.router.match('HEAD', path):
                return json_response(405, {"error": f"方法 {method} 不允许"})
            return json_response(404, {"error": f"路径 '{path}' 不存在"})

        _, handler = match

        # 执行中间件
        for mw in self.middleware:
            mw(req)

        # 执行处理器
        try:
            status, data = handler(req, req['body'])
            if isinstance(data, dict) or isinstance(data, list):
                return json_response(status, data)
            return build_response(status, data if isinstance(data, bytes) else str(data).encode())
        except Exception as e:
            return json_response(500, {"error": f"服务器内部错误: {str(e)}"})


# ──── 服务器循环 ────

class MiniHTTPServer:
    """迷你 HTTP 服务器"""

    def __init__(self, host: str = '127.0.0.1', port: int = 8080, app: MiniApp = None):
        self.host = host
        self.port = port
        self.app = app or MiniApp()
        self.running = False

    def start(self):
        self.running = True
        server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_sock.bind((self.host, self.port))
        server_sock.listen(128)
        server_sock.settimeout(1.0)  # 1 秒超时允许检测 running 标志

        print(f"🚀 MiniHTTPD 启动在 http://{self.host}:{self.port}")
        print(f"   按 Ctrl+C 停止")

        while self.running:
            try:
                conn, addr = server_sock.accept()
            except socket.timeout:
                continue
            except KeyboardInterrupt:
                break

            try:
                conn.settimeout(5)
                data = conn.recv(65536)
                if data:
                    response = self.app.handle_request(data)
                    conn.sendall(response)
            except Exception as e:
                print(f"   处理请求时出错: {e}")
            finally:
                try:
                    conn.close()
                except Exception:
                    pass

        server_sock.close()
        print("🛑 服务器已停止")

    def stop(self):
        self.running = False


# ──── 入口 ────

if __name__ == '__main__':
    import sys

    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
    server = MiniHTTPServer(port=port)

    # 优雅退出
    def signal_handler(sig, frame):
        print("\n收到停止信号...")
        server.stop()

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    server.start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284

测试你的迷你服务器

bash

# 终端 1：启动服务器
python3 mini_httpd.py 8080

# 终端 2：发请求
# GET
curl -v http://localhost:8080/
curl -v http://localhost:8080/api/items

# POST（创建）
curl -v -X POST http://localhost:8080/api/items \
  -H "Content-Type: application/json" \
  -d '{"name": "笔记本", "price": 2999}'

# GET（验证创建）
curl -s http://localhost:8080/api/items | python3 -m json.tool

# PUT（替换）
curl -v -X PUT http://localhost:8080/api/items \
  -H "Content-Type: application/json" \
  -d '[{"name": "新笔记本", "price": 3999}]'

# DELETE
curl -v -X DELETE http://localhost:8080/api/items

# 404 / 405
curl -v http://localhost:8080/nonexistent
curl -v -X DELETE http://localhost:8080/

常见陷阱

实现一个 HTTP 调试代理——转发请求并打印详细日志：

python

#!/usr/bin/env python3
"""mini_proxy.py — 调试用 HTTP 代理，打印所有请求/响应细节"""
import socket
import sys
from urllib.parse import urlparse

def proxy_request(client_sock: socket.socket, data: bytes):
    """解析请求，转发，回传响应"""
    # 解析请求行
    lines = data.split(b'\r\n')
    first_line = lines[0].decode()
    method, url, version = first_line.split(' ', 2)

    # 提取 Host（从 URL 或 Header）
    parsed = urlparse(url)
    host = parsed.hostname or 'localhost'
    port = parsed.port or 80
    path = parsed.path or '/'
    if parsed.query:
        path += '?' + parsed.query

    # 🎯 日志：请求
    print(f"\n>>> {method} {url}")
    for line in lines[1:]:
        if line:
            print(f">>>   {line.decode(errors='replace')}")

    # 连接到目标服务器
    try:
        remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        remote.settimeout(10)
        remote.connect((host, port))

        # 重写请求行（用 path 代替 full URL）
        new_req = f"{method} {path} {version}\r\n".encode()
        # 保留其他 headers（跳过原来的请求行）
        new_req += b'\r\n'.join(lines[1:])
        new_req += b'\r\n'

        remote.sendall(new_req)

        # 读取响应并回传
        response = b''
        while True:
            chunk = remote.recv(65536)
            if not chunk:
                break
            response += chunk

        # 🎯 日志：响应状态行
        status_line = response.split(b'\r\n')[0].decode()
        print(f"<<< {status_line}")
        print(f"<<< 响应大小: {len(response)} bytes")

        client_sock.sendall(response)
        remote.close()

    except Exception as e:
        error_resp = f"HTTP/1.1 502 Bad Gateway\r\nContent-Length: {len(str(e))}\r\n\r\n{e}".encode()
        client_sock.sendall(error_resp)
        print(f"<<< 502 Bad Gateway: {e}")

def main(listen_port: int = 8888):
    proxy = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    proxy.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    proxy.bind(('127.0.0.1', listen_port))
    proxy.listen(5)
    print(f"🔍 HTTP 调试代理运行在 127.0.0.1:{listen_port}")
    print(f"   用法: curl -x http://127.0.0.1:{listen_port} http://example.com")

    try:
        while True:
            conn, addr = proxy.accept()
            data = conn.recv(65536)
            if data:
                proxy_request(conn, data)
            conn.close()
    except KeyboardInterrupt:
        print("\n代理已停止")
    finally:
        proxy.close()

if __name__ == '__main__':
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8888
    main(port)

通关挑战

基础题（30 分）

请求解析：从原始 bytes b"GET /search?q=http HTTP/1.1\r\nHost: example.com\r\n\r\n" 中提取方法、路径、查询参数。
状态码匹配：为以下场景选择正确的 HTTP 状态码：
- 用户尝试访问 /admin 但没有登录
- 用户尝试 PUT 一个不存在的资源
- 服务器数据库挂了，但能正常返回
- 客户端发请求太频繁
Header 识别：说明 Cache-Control: no-cache 与 Cache-Control: no-store 的根本区别。

进阶题（40 分）

写响应生成器：用 Python 生成一个响应，包含以下 headers：Content-Type: application/json、Cache-Control: public, max-age=3600、Access-Control-Allow-Origin: *，Body 为 {"status": "ok"}。
Cookie 管理器：写一个简单的类，解析 Set-Cookie 响应头，存储 cookie，并在后续请求中自动附加 Cookie 头。

挑战题（30 分）

增强迷你服务器：给 MiniHTTPServer 增加：
- Transfer-Encoding: chunked 支持（流式响应）
- 静态文件服务（从 ./static/ 目录自动映射）
- 支持 If-None-Match（ETag 条件请求）

验收标准

[ ] 能从原始 bytes 解析任意 HTTP/1.1 请求，提取方法/路径/headers/body
[ ] 能手动构建一个合法的 HTTP 响应 bytes
[ ] mini_httpd.py 能启动并对 curl 返回正确的响应
[ ] 能区分安全方法、幂等方法并举例
[ ] 能解释 Cache-Control 各个指令的效果
[ ] 能解释 CORS 预检请求和简单请求的区别
[ ] 能说出 HTTP/1.1 持久连接、管线化、HTTP/2 多路复用的演进原因

常见卡点

卡点	原因	解法
\r\n 和 \n 搞混	DevTools 显示友好格式，看不到原始报文	HTTP 标准就是 `\r\n`。用 `curl --raw` 或 nc 看原始
Content-Length 忘记闭合	recv 无限等待或提前截断	一次性 recv 全部后先找 `\r\n\r\n`，再用 Content-Length 切 Body
CORS 以为阻止了请求	浏览器实际上"发出请求但不让你读响应"	DevTools Network 标签看"actual request"发出了，JS 拿不到才是 CORS
no-cache 以为不缓存	名字有误导性	no-cache = 验证后再用缓存；no-store = 完全不存
不知道持久连接何时关闭	觉得连接永远不断	Connection: close 或超时后关闭。超时默认 ~60-120 秒

现在不需要理解

HTTP/2 的 HPACK 头压缩：Huffman 编码 + 静态/动态字典。在你有大规模 API 服务（每分钟数百万请求）之前，不用深究。
WebSocket 升级握手：Upgrade: websocket 的 101 Switching Protocols 响应 + 密钥验证。它是独立协议，HTTP 只负责"牵线"。
HTTP/3（QUIC）：基于 UDP 的 HTTP，解决了 TCP 队头阻塞。等 QUIC 在你的服务器上默认启用再说。
OAuth 2.0 / OpenID Connect：HTTP 协议栈上层的认证授权框架。单独需要一个完整的章节来讲。

旅人笔记

我真正理解 HTTP 是在写这个迷你服务器的过程中。
之前一直用 Flask/FastAPI/Django，觉得 HTTP 就是个 "rest.get() → 200" 的抽象。第一次用 socket.bind + recv 接到原始请求第一行 GET / HTTP/1.1 的时候，才明白框架究竟隐藏了什么。
大概花了半小时查 \r\n 的问题——我的服务器一直收不到请求，直到发现 Python 的 sendall 发送的 \n 和 C 客户端的 \r\n 混用了。这就是为什么 HTTP 标准里一切都是 CRLF，不是 LF 或 CR。
然后写了路由匹配、CORS 头、JSON 编码/解码——才意识到每天用的 Spring Boot / Express / Flask 究竟干了多少你没看见的事。
试着运行你的迷你服务器，然后用同一个 curl 去访问。看到原始报文对上的那一刻，你对 HTTP 的理解就再也不一样了。

→ 下一站预告

第 5 章：DNS 与内容分发网络

URL 背后的"电话本"——DNS 不仅把域名翻译成 IP，它还决定了你的请求去找哪个 CDN 节点。你会用 Python 实现 DNS 解析器，并理解 CDN 如何把 300ms 的延迟压到 30ms。

第4章：HTTP 与 Web 服务器 ​

元数据卡 ​

你在哪 ​

你的任务 ​

遭遇战 → 获得技能 ​

4.1 HTTP 请求：你每天发的"信"长什么样 ​

请求行 ​

Headers（键值对，每行一个） ​

空行 + Body ​

4.2 HTTP 响应：服务器回的信 ​

4.3 HTTP 方法：动词的语义 ​

方法差异的底层逻辑 ​

4.4 状态码：服务器的"脸色" ​

五大家族 ​

关键状态码深度理解 ​

状态码选择速查 ​

4.5 Headers：HTTP 的"元力量" ​

Cache-Control：如何对客户端和代理说"缓存多久" ​

Cookie 与 Set-Cookie ​

CORS：跨域不是"安全问题"，是浏览器帮你做的限制 ​

4.6 HTTP 持久连接与管线化：别为每个请求新建 TCP ​

短连接时代（HTTP/1.0 默认） ​

持久连接（HTTP/1.1 默认） ​

管线化——HTTP/1.1 的尝试与失败 ​

项目实战：写一个迷你 Web 服务器 ​

测试你的迷你服务器 ​

常见陷阱 ​

通关挑战 ​

基础题（30 分） ​

进阶题（40 分） ​

挑战题（30 分） ​

验收标准 ​

常见卡点 ​

现在不需要理解 ​

旅人笔记 ​

→ 下一站预告 ​

第4章：HTTP 与 Web 服务器

元数据卡

你在哪

你的任务

遭遇战 → 获得技能

4.1 HTTP 请求：你每天发的"信"长什么样

请求行

Headers（键值对，每行一个）

空行 + Body

4.2 HTTP 响应：服务器回的信

4.3 HTTP 方法：动词的语义

方法差异的底层逻辑

4.4 状态码：服务器的"脸色"

五大家族

关键状态码深度理解

状态码选择速查

4.5 Headers：HTTP 的"元力量"

Cache-Control：如何对客户端和代理说"缓存多久"

Cookie 与 Set-Cookie

CORS：跨域不是"安全问题"，是浏览器帮你做的限制

4.6 HTTP 持久连接与管线化：别为每个请求新建 TCP

短连接时代（HTTP/1.0 默认）

持久连接（HTTP/1.1 默认）

管线化——HTTP/1.1 的尝试与失败

项目实战：写一个迷你 Web 服务器

测试你的迷你服务器

常见陷阱

通关挑战

基础题（30 分）

进阶题（40 分）

挑战题（30 分）

验收标准

常见卡点

现在不需要理解

旅人笔记

→ 下一站预告