Python socket 實戰：TCP client/server、timeout 與簡易通訊協定

一. 前言：HTTP 很方便，但底下其實是 byte stream

#

平常寫 Python 網路程式，我們很少直接碰 socket。 如果要打 API，用 httpx 或 requests。 如果要寫 server，用 FastAPI、Django、Flask。 如果要連資料庫、Redis、message queue，通常也有現成 client。 這樣很好。拍拍君不是要你回到石器時代，每天自己手刻 HTTP parser。 但懂一點 socket 很有用。 它會讓你理解幾件平常常踩到的問題：

• 為什麼 recv() 不一定一次拿到完整訊息
• 為什麼 timeout 要分成 connect timeout 和 read timeout
• 為什麼 TCP 沒有所謂「一包 message」
• 為什麼 server 要處理半斷線和空 bytes
• 為什麼 protocol framing 比 send() 更重要 今天這篇不是要做 production-grade server。 我們會用標準庫 socket 寫一個很小的 TCP client/server，然後一步步加上：
• timeout
• line-based protocol
• length-prefixed protocol
• 簡單 echo server 看完以後，你不一定會直接拿 socket 寫服務，但你會更懂上層工具為什麼要那樣設計。 這種知識有點像知道咖啡機怎麼加壓。平常不必拆機，但出問題時你會比較不慌。

二. 安裝：標準庫內建，先開一個練習專案

#

socket 是 Python 標準庫。不用安裝。 但我們還是用 uv 開一個乾淨目錄，方便放範例：

uv init socket-lab
cd socket-lab

今天會建立幾個檔案：

socket-lab/
├── echo_server.py
├── echo_client.py
├── line_server.py
├── line_client.py
├── length_protocol.py

如果你不想用 uv，直接用系統 Python 也可以：

python3 echo_server.py

拍拍君建議先在本機練習，host 都用 127.0.0.1。 127.0.0.1 是 loopback address，意思是「連回自己這台機器」。 不要一開始就開到公開網路。網路程式很可愛，但直接裸奔到外面就不可愛了。

三. 第一個 TCP server：listen、accept、recv、sendall

#

先建立 echo_server.py：

from __future__ import annotations
import socket
HOST = "127.0.0.1"
PORT = 8765
def main() -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((HOST, PORT))
        server.listen()
        print(f"listening on {HOST}:{PORT}")
        conn, addr = server.accept()
        with conn:
            print(f"connected by {addr}")
            data = conn.recv(1024)
            print(f"received: {data!r}")
            conn.sendall(data)
if __name__ == "__main__":
    main()

這裡有幾個核心概念。 socket.AF_INET 代表 IPv4。 socket.SOCK_STREAM 代表 TCP。 bind() 把 server 綁到本機 address 和 port。 listen() 開始接受連線。 accept() 會等 client 連進來，回傳一個新的 connection socket 和 client address。 注意：server socket 負責聽門鈴，conn socket 才是和某個 client 對話的連線。 先開一個終端機跑 server：

python echo_server.py

它會停在 accept() 等 client。 這是正常的。不是卡住，只是很有耐心。

四. 第一個 TCP client：connect、sendall、recv

#

另開一個終端機，建立 echo_client.py：

from __future__ import annotations
import socket
HOST = "127.0.0.1"
PORT = 8765
def main() -> None:
    message = "hello from pypy\n".encode("utf-8")
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.connect((HOST, PORT))
        sock.sendall(message)
        data = sock.recv(1024)
    print(f"server replied: {data.decode('utf-8')}")
if __name__ == "__main__":
    main()

執行：

python echo_client.py

你應該會看到 client 收到原本送出的字串。 server 那邊會印出 bytes：

received: b'hello from pypy\n'

這裡第一個重要觀念出現了： socket 傳的是 bytes，不是 Python string。 所以送出前要 encode()，收到後要 decode()。 sendall() 和 send() 也不一樣。 send() 可能只送出一部分 bytes，回傳實際送出的長度。 sendall() 會幫你努力送完整段資料，除非中途出錯。 初學時，優先用 sendall()。 不要跟 partial write 賭運氣。賭輸時 bug 很難看。

五. TCP 是 byte stream，不是 message stream

#

很多人第一次寫 socket 會以為：

sock.sendall(b"hello")
sock.sendall(b"world")

對方就會收到兩次：

hello
world

不一定。 TCP 只保證 bytes 順序正確、可靠送達。 它不保證你每次 sendall() 都會對應到對方一次 recv()。 對方可能收到：

helloworld

也可能先收到：

hel

下一次才收到：

loworld

這不是 Python 壞掉。這是 TCP 的本質。 所以你不能把 recv(1024) 當作「讀一則訊息」。 你只能把它當作「從 stream 裡拿一些 bytes」。 真正的訊息邊界，要由你的 protocol 定義。 常見方法有兩種：

• 用換行符號當結尾：line-based protocol
• 先傳長度，再傳內容：length-prefixed protocol 我們兩種都做一次。

六. Line protocol：用 \n 當訊息邊界

#

很多文字型 protocol 都是 line-based。 例如簡化想像下，你可以把一個命令寫成：

PING
ECHO hello
QUIT

每一行是一則訊息。 建立 line_server.py：

from __future__ import annotations
import socket
HOST = "127.0.0.1"
PORT = 8766
def handle_line(line: str) -> str:
    command = line.strip()
    if command == "PING":
        return "PONG\n"
    if command.startswith("ECHO "):
        return command.removeprefix("ECHO ") + "\n"
    if command == "QUIT":
        return "BYE\n"
    return "ERR unknown command\n"
def main() -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((HOST, PORT))
        server.listen()
        print(f"line server on {HOST}:{PORT}")
        conn, addr = server.accept()
        with conn, conn.makefile("rwb") as stream:
            print(f"connected by {addr}")
            for raw_line in stream:
                line = raw_line.decode("utf-8")
                reply = handle_line(line)
                stream.write(reply.encode("utf-8"))
                stream.flush()
                if line.strip() == "QUIT":
                    break
if __name__ == "__main__":
    main()

這裡用了 conn.makefile("rwb")。 它把 socket 包成 file-like object，讓你可以用 for raw_line in stream 一行一行讀。 這對 line protocol 很方便。 再寫 line_client.py：

from __future__ import annotations
import socket
HOST = "127.0.0.1"
PORT = 8766
def send_command(stream, command: str) -> str:
    stream.write((command + "\n").encode("utf-8"))
    stream.flush()
    return stream.readline().decode("utf-8").strip()
def main() -> None:
    with socket.create_connection((HOST, PORT), timeout=3) as sock:
        with sock.makefile("rwb") as stream:
            print(send_command(stream, "PING"))
            print(send_command(stream, "ECHO 拍拍君正在測 socket"))
            print(send_command(stream, "WHAT"))
            print(send_command(stream, "QUIT"))
if __name__ == "__main__":
    main()

執行順序：

python line_server.py

另一個終端機：

python line_client.py

你會看到：

PONG
拍拍君正在測 socket
ERR unknown command
BYE

line protocol 的優點是簡單、好 debug。 缺點是內容如果本身可能有換行，就要 escape 或改用別的 framing。

七. Length-prefixed protocol：先說長度，再送內容

#

如果訊息是 JSON、binary data，或內容可能包含換行，length-prefix 通常比較穩。 概念是： 先用固定長度的 header 表示 body 幾個 bytes。 例如前 4 bytes 是 unsigned integer，接著才是 payload。 建立 length_protocol.py：

from __future__ import annotations
import json
import socket
import struct
from typing import Any
HEADER_SIZE = 4
def recv_exactly(sock: socket.socket, size: int) -> bytes:
    chunks: list[bytes] = []
    remaining = size
    while remaining > 0:
        chunk = sock.recv(remaining)
        if chunk == b"":
            raise ConnectionError("socket closed before receiving enough data")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)
def send_message(sock: socket.socket, payload: dict[str, Any]) -> None:
    body = json.dumps(payload).encode("utf-8")
    header = struct.pack("!I", len(body))
    sock.sendall(header + body)
def recv_message(sock: socket.socket) -> dict[str, Any]:
    header = recv_exactly(sock, HEADER_SIZE)
    (body_size,) = struct.unpack("!I", header)
    body = recv_exactly(sock, body_size)
    return json.loads(body.decode("utf-8"))

struct.pack("!I", len(body)) 的意思是：

• !：network byte order，也就是 big-endian
• I：4-byte unsigned integer 這樣 header 永遠是 4 bytes。 接收端先讀 4 bytes，知道 body 長度，再精準讀那麼多 bytes。 recv_exactly() 是這段的重點。 它處理了 partial read。 如果你只寫：

body = sock.recv(body_size)

那你其實是在假設「一次 recv 就會拿完」。 小範例可能可以。真實網路上不保證。 拍拍君對這種「今天剛好可以」的程式碼比較沒耐心。它不是穩，只是還沒被測到。

八. timeout：不要讓程式永遠等下去

#

網路程式最討厭的一種 bug 是：什麼都沒壞，但程式永遠卡著。 可能是對方 server 沒回。 可能是網路中間斷了。 可能是 protocol 雙方都在等對方先講話。 所以 timeout 很重要。 最簡單的方式是用 socket.create_connection()：

import socket
with socket.create_connection(("example.com", 80), timeout=3) as sock:
    sock.sendall(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
    sock.settimeout(3)
    data = sock.recv(4096)
print(data[:200])

create_connection(..., timeout=3) 主要控制連線階段。 sock.settimeout(3) 會影響後續 socket operation，例如 recv() 和 send()。 真實應用裡，connect timeout 和 read timeout 常常要分開思考。 例如：

• connect timeout：1 到 3 秒
• read timeout：10 到 30 秒
• long polling 或 streaming：更長，甚至特別處理 如果你之前用過 httpx.Timeout，就會知道它把 connect、read、write、pool 分開。 底層原因就在這裡。 上層工具不是故意把設定做複雜。網路本來就有不同階段。

九. recv() 回傳空 bytes 代表什麼？

#

這行很重要：

chunk = sock.recv(4096)

如果 chunk == b""，代表對方已經關閉連線。 不是「收到空訊息」。 是「這條 stream 結束了」。 所以常見讀取迴圈會長這樣：

def read_until_closed(sock: socket.socket) -> bytes:
    chunks: list[bytes] = []
    while True:
        chunk = sock.recv(4096)
        if chunk == b"":
            break
        chunks.append(chunk)
    return b"".join(chunks)

但不是所有 protocol 都適合讀到對方關閉。 HTTP/1.1、長連線、聊天服務、資料庫連線，通常一條 connection 會承載多個 request/response。 這時你更需要明確 framing，不能只靠「對方關閉」當訊息結束。 如果 client 等 server 關閉，server 又等 client 下一個命令，恭喜，兩邊就會一起安靜地發呆。 這種 bug 沒有戲劇效果。只有 timeout 之後的尷尬。

十. 常見錯誤與拍拍君的碎念

#

第一，不要忘記處理 bytes/string。 socket 收送的是 bytes。如果 protocol 是文字，明確指定 encoding。

payload = text.encode("utf-8")
text = payload.decode("utf-8")

第二，不要假設一次 recv() 就是一則訊息。 這是 socket 初學者最常見的坑。 你要有 protocol。 換行也好，長度 header 也好，總之要定義邊界。 第三，不要讓程式永遠等。 加 timeout。 即使只是內部工具，也要讓失敗有機會變成錯誤訊息，而不是變成一個永遠不動的終端機。 第四，不要一開始就開公開 host。 練習時用：

HOST = "127.0.0.1"

如果你改成：

HOST = "0.0.0.0"

意思是讓 server 聽所有網路介面。 這在 Docker 或內網服務很常見，但也表示你的服務可能被其他機器連到。 該加防火牆、認證和部署邊界時，不要裝沒看到。 第五，不要自己重寫成熟 protocol。 想打 HTTP API，用 httpx。 想寫 web API，用 FastAPI。 想要 TLS、proxy、redirect、cookie、compression，請交給成熟工具。 學 socket 是為了理解底層，不是為了把所有輪子都重刻一次。 拍拍君很喜歡理解底層，但不喜歡無謂受苦。

結語

#

今天我們用 Python 標準庫 socket 走過 TCP 的基本流程。 你看到了：

• socket()、bind()、listen()、accept() 的 server 流程
• connect()、sendall()、recv() 的 client 流程
• TCP 是 byte stream，不是 message stream
• line protocol 和 length-prefixed protocol 的差別
• timeout 為什麼重要
• recv() 回傳空 bytes 代表連線關閉 如果你平常只用高階 HTTP client，這篇可能有一點底層。 但這種底層感很值得保留一點。 下次 API request 卡住、server 沒回應、Docker service port 打不通，你會比較知道該問什麼問題。 不是每個人都需要每天寫 socket。 但每個寫網路程式的人，都值得知道 socket 在下面默默做什麼。

延伸閱讀

#

• Python 官方文件：socket
• Python 官方文件：socketserver
• 拍拍君：Python HTTPX 實戰
• 拍拍君：FastAPI + Streamlit 實戰