实战：手写极简 HTTP 服务器与客户端

python3 -c '
import socket, threading, json
 
def handle(conn):
    try:
        # ── 1. 读到头部结束 ──────────────────────────
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = conn.recv(4096)
            if not chunk:
                return
            buf += chunk
        head, _, rest = buf.partition(b"\r\n\r\n")
        lines = head.decode("iso-8859-1").split("\r\n")
 
        # ── 2. 解析请求行与头部 ──────────────────────
        method, path, version = lines[0].split(" ", 2)
        headers = {}
        for line in lines[1:]:
            k, _, v = line.partition(":")
            headers[k.strip().lower()] = v.strip()
 
        # ── 3. 按 Content-Length 读完请求体 ──────────
        need = int(headers.get("content-length", 0))
        body = rest
        while len(body) < need:
            body += conn.recv(4096)
 
        # ── 4. 路由 ─────────────────────────────────
        if path == "/hello":
            status, ctype, out = "200 OK", "text/plain; charset=utf-8", b"Hello, HTTP!\n"
        elif path == "/headers":
            out = json.dumps(headers, ensure_ascii=False, indent=1).encode()
            status, ctype = "200 OK", "application/json"
        elif path == "/echo" and method == "POST":
            status, ctype, out = "200 OK", "application/octet-stream", body
        else:
            status, ctype, out = "404 Not Found", "text/plain", b"no such route\n"
 
        # ── 5. 拼响应报文 ────────────────────────────
        resp = ("HTTP/1.1 " + status + "\r\n"
                "Content-Type: " + ctype + "\r\n"
                "Content-Length: " + str(len(out)) + "\r\n"
                "Server: handmade/0.1\r\n"
                "Connection: close\r\n\r\n").encode() + out
        conn.sendall(resp)
    finally:
        conn.close()
 
def serve():
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", 8000))
    srv.listen(16)
    while True:
        c, _ = srv.accept()
        threading.Thread(target=handle, args=(c,), daemon=True).start()
 
threading.Thread(target=serve, daemon=True).start()
import time; time.sleep(0.3)
 
import subprocess
subprocess.run(["curl", "-s", "http://127.0.0.1:8000/hello"], check=False)
subprocess.run(["curl", "-s", "http://127.0.0.1:8000/headers", "-H", "X-Course: http"], check=False)
print()
subprocess.run(["curl", "-s", "-X", "POST", "http://127.0.0.1:8000/echo", "-d", "round-trip!"], check=False)
print()
subprocess.run(["curl", "-s", "-o", "/dev/null", "-w", "未知路径 -> %{http_code}\n",
                "http://127.0.0.1:8000/nope"], check=False)
'

echo "served by stdlib server" > data.txt
python3 -m http.server 8000 --bind 127.0.0.1 >/dev/null 2>&1 &
for _i in $(seq 1 50); do (exec 3<>/dev/tcp/127.0.0.1/8000) 2>/dev/null && break; sleep 0.1; done
 
python3 -c '
import socket
 
def http_get(host, port, path):
    s = socket.create_connection((host, port), timeout=5)
    req = ("GET " + path + " HTTP/1.1\r\n"
           "Host: " + host + "\r\n"
           "User-Agent: handmade-client/0.1\r\n"
           "Connection: close\r\n\r\n")
    s.sendall(req.encode())
 
    # 收齐头部
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = s.recv(4096)
        if not chunk:
            break
        buf += chunk
    head, _, body = buf.partition(b"\r\n\r\n")
    lines = head.decode("iso-8859-1").split("\r\n")
 
    version, code, reason = lines[0].split(" ", 2)
    headers = {}
    for line in lines[1:]:
        k, _, v = line.partition(":")
        headers[k.strip().lower()] = v.strip()
 
    # 按 Content-Length 读完剩余的体
    need = int(headers.get("content-length", 0))
    while len(body) < need:
        chunk = s.recv(4096)
        if not chunk:
            break
        body += chunk
    s.close()
    return int(code), reason, headers, body
 
code, reason, headers, body = http_get("127.0.0.1", 8000, "/data.txt")
print("状态:", code, reason)
print("Content-Type:", headers.get("content-type"))
print("Content-Length:", headers.get("content-length"), "实收:", len(body))
print("体:", body.decode().strip())
 
code, reason, _, _ = http_get("127.0.0.1", 8000, "/missing.txt")
print("再试一个不存在的文件:", code, reason)
'

python3 -c '
import socket, threading, time
 
# ── 迷你服务器：只有一个路由，但支持 keep-alive ──
def handle(conn):
    while True:                          # 循环处理同一连接上的多个请求
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = conn.recv(4096)
            if not chunk:
                conn.close(); return
            buf += chunk
        req_line = buf.split(b"\r\n", 1)[0].decode()
        out = ("你请求了: " + req_line + "\n").encode()
        conn.sendall(("HTTP/1.1 200 OK\r\n"
                      "Content-Length: " + str(len(out)) + "\r\n"
                      "\r\n").encode() + out)   # 不发 Connection: close -> 保持连接
 
def serve():
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", 8000)); srv.listen(4)
    while True:
        c, _ = srv.accept()
        threading.Thread(target=handle, args=(c,), daemon=True).start()
 
threading.Thread(target=serve, daemon=True).start()
time.sleep(0.2)
 
# ── 迷你客户端：一条连接串行发 3 个请求（验证 keep-alive）──
s = socket.create_connection(("127.0.0.1", 8000))
for path in ("/first", "/second", "/third"):
    s.sendall(("GET " + path + " HTTP/1.1\r\nHost: local\r\n\r\n").encode())
    buf = b""
    while b"\r\n\r\n" not in buf:
        buf += s.recv(4096)
    head, _, body = buf.partition(b"\r\n\r\n")
    need = 0
    for line in head.split(b"\r\n"):
        if line.lower().startswith(b"content-length"):
            need = int(line.split(b":")[1])
    while len(body) < need:
        body += s.recv(4096)
    print(body.decode().strip(), "| 复用同一条连接:", s.getsockname())
s.close()
'