Python socket：TCP 字节流、地址解析与平台接口

学习教程更新于 2026-10-03编辑本文

socket 提供操作系统套接字接口。TCP 服务端先创建监听套接字，依次调用 bind()、listen()、accept()；真正的收发发生在 accept() 返回的连接套接字上。客户端创建套接字后调用 connect()。这与 UDP 按数据报收发的模型不同。

以下示例以 Python 3.11 为基准。前四个程序是只服务一个客户端的 TCP 回显实验，默认使用本机地址和端口 50007；每次阻塞操作设定五秒超时。Windows 原始 IPv4 套接字和 Linux CAN 位于后面的独立程序中，不能当作所有系统都支持的通用接口。Python socket 文档

TCP 是字节流，单次 recv 不代表完整消息

recv(1024) 表示一次最多读取 1024 字节，可能只得到当前可用的一部分。连接正常结束时，返回空字节串 b''。send() 返回本次实际送出的字节数；这里用 sendall() 完成整个缓冲区的发送，但它仍不表示对方应用已经处理了数据。

实验协议是：客户端发送数据，调用 shutdown(SHUT_WR) 关闭自己的发送方向，然后继续读取；服务端回显所有收到的数据，读到 EOF 后关闭连接；客户端读到服务端 EOF 后拼接结果。这种半关闭方式让两端都知道何时结束，适合本例的一次性交换。持续复用的连接通常需要长度前缀、分隔符等消息边界。

IPv4 服务端：echo_ipv4_server.py

先运行服务端，再运行客户端。默认 127.0.0.1 只监听本机；原文的空字符串 '' 表示 IPv4 通配地址，会监听所有可用本地 IPv4 接口，不应把二者混为一谈。--port 0 可让系统分配空闲端口，程序会打印实际端口。

import argparse
import os
import socket


def serve(host="127.0.0.1", port=50007):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
        if os.name == "posix":
            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listener.settimeout(5.0)
        listener.bind((host, port))
        listener.listen(1)
        print("LISTENING", listener.getsockname()[1], flush=True)
        conn, address = listener.accept()
        with conn:
            conn.settimeout(5.0)
            print("Connected by", address, flush=True)
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                conn.sendall(data)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=50007)
    args = parser.parse_args()
    serve(args.host, args.port)

IPv4 客户端：echo_ipv4_client.py

在另一个终端运行客户端；指定端口时与服务端保持一致。这里连接本机服务，不依赖原文示例中的外部主机 daring.cwi.nl 是否存在或开放端口。

import argparse
import socket


def exchange(host="127.0.0.1", port=50007, payload=b"Hello, world"):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as conn:
        conn.settimeout(5.0)
        conn.connect((host, port))
        conn.sendall(payload)
        conn.shutdown(socket.SHUT_WR)
        chunks = []
        while True:
            data = conn.recv(1024)
            if not data:
                break
            chunks.append(data)
        return b"".join(chunks)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=50007)
    args = parser.parse_args()
    result = exchange(args.host, args.port)
    assert result == b"Hello, world"
    print("Received", repr(result))

两个文件分别运行后，客户端输出 Received b'Hello, world'，服务端完成这一个连接后退出。要服务多个客户端，需要另行设计重复接收、并发与关闭机制；只把 listen(1) 的参数调大，不会自动创建多个处理线程。

通过 getaddrinfo 尝试 IPv4 或 IPv6

AF_UNSPEC 允许解析器返回不同地址族，getaddrinfo() 给出创建套接字所需的地址族、类型、协议和地址。下面的服务端依次尝试候选项，并保留第一个成功监听的套接字；客户端依次尝试连接候选项。失败的候选套接字也会关闭。

这不等于同时建立 IPv4 和 IPv6 两个监听器。一个 IPv6 套接字是否接收 IPv4 映射地址，与平台和 IPV6_V6ONLY 设置有关。需要明确的双栈服务时，可设计两个监听器，或检查平台能力后使用 socket.create_server(..., family=AF_INET6, dualstack_ipv6=True)。地址解析和双栈辅助函数

地址族无关服务端：echo_any_server.py

默认主机为 localhost。传入 --host "*" 时，本程序把它转换为 None 并配合 AI_PASSIVE 请求通配监听地址；仍然只选第一个成功的地址族。

import argparse
import os
import socket


def open_listener(host, port):
    flags = socket.AI_PASSIVE if host is None else 0
    candidates = socket.getaddrinfo(
        host, port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, flags
    )
    last_error = None
    for family, kind, protocol, _, address in candidates:
        listener = None
        try:
            listener = socket.socket(family, kind, protocol)
            if os.name == "posix":
                listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            listener.settimeout(5.0)
            listener.bind(address)
            listener.listen(1)
            return listener
        except OSError as exc:
            last_error = exc
            if listener is not None:
                listener.close()
    raise OSError("could not open a listening socket") from last_error


def serve(host="localhost", port=50007):
    with open_listener(host, port) as listener:
        print("LISTENING", listener.getsockname()[1], flush=True)
        conn, address = listener.accept()
        with conn:
            conn.settimeout(5.0)
            print("Connected by", address, flush=True)
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                conn.sendall(data)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="localhost")
    parser.add_argument("--port", type=int, default=50007)
    args = parser.parse_args()
    serve(None if args.host == "*" else args.host, args.port)

地址族无关客户端：echo_any_client.py

import argparse
import socket


def open_connection(host, port):
    last_error = None
    for family, kind, protocol, _, address in socket.getaddrinfo(
        host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
    ):
        conn = None
        try:
            conn = socket.socket(family, kind, protocol)
            conn.settimeout(5.0)
            conn.connect(address)
            return conn
        except OSError as exc:
            last_error = exc
            if conn is not None:
                conn.close()
    raise OSError("could not connect to any resolved address") from last_error


def exchange(host="localhost", port=50007, payload=b"Hello, world"):
    with open_connection(host, port) as conn:
        conn.sendall(payload)
        conn.shutdown(socket.SHUT_WR)
        chunks = []
        while True:
            data = conn.recv(1024)
            if not data:
                break
            chunks.append(data)
        return b"".join(chunks)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="localhost")
    parser.add_argument("--port", type=int, default=50007)
    args = parser.parse_args()
    result = exchange(args.host, args.port)
    assert result == b"Hello, world"
    print("Received", repr(result))

需要定向实验时，在这两个程序上同时指定 --host 127.0.0.1 或 --host ::1。IPv6 回环是否可用应以本机实际环境为准，不能仅凭 AF_UNSPEC 断言通信已经同时经过两种协议。

SO_REUSEADDR 不是端口冲突的通用解法

Address already in use 可能来自仍在监听的进程，也可能与先前连接的地址重用规则有关。TIME_WAIT 是 TCP 连接状态；并不是所有端口占用都由它引起。

上面两个服务端在 POSIX 系统上、调用 bind() 之前设置 SO_REUSEADDR，保留原例的设置顺序。其具体语义依平台而异，也不能保证任意仍被使用的地址都能重新绑定。Windows 的 SO_REUSEADDR 允许的地址重用行为与 POSIX 不同，本例不在 Windows 上盲目启用；需要独占监听时另行了解 SO_EXCLUSIVEADDRUSE。Windows 地址重用选项

Windows 原始 IPv4 接收：单独选择接口

下面的程序需要 Windows、支持的原始套接字能力，以及足够权限。调用时明确传入本机接口的 IPv4 地址，例如 python raw_ipv4_capture.py 192.0.2.10 中的地址必须换成机器实际拥有的地址；不再用主机名解析结果猜测“公共接口”。这里只接收一个 IPv4 数据包，最多等待五秒，然后关闭接收模式并释放套接字。

SIO_RCVALL 面向绑定接口上的 IP 流量，不会把原始 IP 套接字变成能抓取任意以太网帧的工具。原文的 IP_HDRINCL 选项也保留在代码中，但它控制发送时是否由调用方提供 IPv4 头，不是启用接收 IP 头的开关。Microsoft SIO_RCVALL

import argparse
import ipaddress
import socket
import sys
import warnings


def capture_one(host):
    if sys.platform != "win32" or not hasattr(socket, "SIO_RCVALL"):
        raise RuntimeError("this example requires Windows SIO_RCVALL")
    address = str(ipaddress.IPv4Address(host))
    with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP) as conn:
        conn.bind((address, 0))
        conn.settimeout(5.0)
        # Retained from the original; controls outgoing IPv4 headers.
        conn.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        enabled = False
        try:
            conn.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
            enabled = True
            return conn.recvfrom(65535)
        finally:
            if enabled:
                try:
                    conn.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
                except OSError as exc:
                    warnings.warn(f"RCVALL_OFF failed before socket close: {exc}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("interface_ipv4")
    args = parser.parse_args()
    print(capture_one(args.interface_ipv4))

这段程序的实际抓包需要上述平台和权限，本次整理只检查了程序语法及模拟的成功、超时、清理失败路径，没有执行真实抓包。

Linux CAN：先区分帧格式与套接字协议

SocketCAN 是 Linux 的接口。经典 CAN 数据最多八字节；CAN FD 的数据长度和结构不同。原文的 =IB3x8s 表示一个本机字节序、标准字段大小、显式填充到 16 字节的经典 CAN 结构：四字节 CAN 标识字段、一字节长度、三字节保留区域、八字节数据。它不是 TCP 网络报文常用的网络大端序编码。

下面的编解码器只接受不带标志的 11 位标准数据帧，不支持扩展 ID、远程帧、错误帧和 CAN FD，也不保存可选的 len8_dlc 信息。原始 can_id 中还可能携带这些标志，不能一律当成纯编号。Linux SocketCAN 文档

可离线运行的经典 CAN 编解码器：can_codec.py

import struct

can_frame_fmt = "=IB3x8s"
can_frame_size = struct.calcsize(can_frame_fmt)


def build_can_frame(can_id, data):
    if type(can_id) is not int or not 0 <= can_id <= 0x7FF:
        raise ValueError("expected an 11-bit standard CAN identifier")
    if not isinstance(data, bytes):
        raise TypeError("data must be bytes")
    if len(data) > 8:
        raise ValueError("classic CAN data must contain at most eight bytes")
    return struct.pack(can_frame_fmt, can_id, len(data), data.ljust(8, b"\x00"))


def dissect_can_frame(frame):
    if not isinstance(frame, bytes):
        raise TypeError("frame must be bytes")
    if len(frame) != can_frame_size:
        raise ValueError("expected exactly one 16-byte classic CAN frame")
    can_id, can_dlc, data = struct.unpack(can_frame_fmt, frame)
    if can_id > 0x7FF or can_dlc > 8:
        raise ValueError("unsupported CAN flags, identifier, or data length")
    return can_id, can_dlc, data[:can_dlc]


if __name__ == "__main__":
    frame = build_can_frame(0x01, b"\x01\x02\x03")
    assert can_frame_size == 16
    assert dissect_can_frame(frame) == (0x01, 3, b"\x01\x02\x03")
    print(dissect_can_frame(frame))

CAN_RAW 收发：can_raw_once.py

与 can_codec.py 保存到同一目录。示例要求已有可用的 vcan0 接口；它不会自行创建、启用或修改网络接口。程序收到一个支持的经典数据帧后，发送重新编码的回显帧以及原例的 0x01、三字节数据帧，然后退出，避免原来的无限回显循环。每次发送都检查完整帧长度。

import argparse
import socket
import sys

from can_codec import build_can_frame, can_frame_size, dissect_can_frame


def exchange_once(interface="vcan0"):
    if not sys.platform.startswith("linux") or not hasattr(socket, "AF_CAN"):
        raise RuntimeError("CAN_RAW requires Linux SocketCAN")
    with socket.socket(socket.AF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as conn:
        conn.settimeout(5.0)
        conn.bind((interface,))
        frame, address = conn.recvfrom(can_frame_size)
        can_id, can_dlc, data = dissect_can_frame(frame)
        print("Received from", address, "can_id=%x can_dlc=%x data=%r" % (
            can_id, can_dlc, data
        ))
        for outgoing in (build_can_frame(can_id, data),
                         build_can_frame(0x01, b"\x01\x02\x03")):
            if conn.send(outgoing) != len(outgoing):
                raise OSError("incomplete CAN frame send")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--interface", default="vcan0")
    args = parser.parse_args()
    exchange_once(args.interface)

CAN_BCM 使用另一种消息协议：can_bcm_open.py

原文的 socket(AF_CAN, SOCK_DGRAM, CAN_BCM) 是广播管理器协议，用 connect((interface,)) 选择 CAN 接口。它的发送消息包含 bcm_msg_head 和相应帧列表，不能直接把上面一个 16 字节 RAW 帧当成完整 BCM 消息。下面仅完整展示打开、连接和关闭，没有编造 BCM 定时发送任务。

import argparse
import socket
import sys


def open_and_close(interface="vcan0"):
    if not sys.platform.startswith("linux") or not hasattr(socket, "CAN_BCM"):
        raise RuntimeError("CAN_BCM requires Linux SocketCAN")
    with socket.socket(socket.AF_CAN, socket.SOCK_DGRAM, socket.CAN_BCM) as conn:
        conn.settimeout(5.0)
        conn.connect((interface,))
        print("Connected CAN_BCM to", interface)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--interface", default="vcan0")
    args = parser.parse_args()
    open_and_close(args.interface)

CAN 编解码器已做离线执行验证；这里的 CAN_RAW、CAN_BCM 连接过程只做了模拟生命周期检查，未在 Linux CAN 接口或实际总线上收发。权限、接口配置与对端行为仍应在目标环境验证。