Skip to main content

socket

socket 模块提供了访问 BSD 套接字? 的接口，是所有现代 Unix 系统、Windows、macOS 和其他一些平台上进行网络编程的基础。

套接字协议族​

根据系统以及构建选项，此模块提供了各种套接字协议簇。常用的协议族包括：

协议族常量地址格式说明
IPv4AF_INET(host, port) 元组
host: 主机名或 IPv4 地址字符串
port: 整数端口号
特殊值: '' 表示所有接口，'<broadcast>' 表示广播地址		用于 IPv4 网络通信
IPv6AF_INET6(host, port, flowinfo, scope_id) 四元组
host: IPv6 地址字符串
port: 整数端口号
flowinfo: 流信息（通常为 0）
scope_id: 作用域 ID（通常为 0）
简化形式: (host, port)		用于 IPv6 网络通信
Unix域AF_UNIX文件系统路径字符串
例如: '/tmp/mysocket.sock'		用于同一台机器上的进程间通信
原始套接字AF_PACKET接口名称字符串
例如: 'eth0'		用于底层网络包访问（Linux）
蓝牙AF_BLUETOOTH蓝牙地址元组，格式取决于协议用于蓝牙通信

地址格式示例​

import socket

# AF_INET 地址示例
address_v4_1 = ('www.example.com', 80)           # 使用主机名
address_v4_2 = ('192.168.1.1', 8080)             # 使用 IPv4 地址
address_v4_3 = ('', 8080)                        # 绑定到所有接口 (INADDR_ANY)
address_v4_4 = ('<broadcast>', 8080)             # 广播地址 (INADDR_BROADCAST)

# AF_INET6 地址示例
address_v6_1 = ('2001:0db8::1', 8080, 0, 0)      # 完整形式
address_v6_2 = ('2001:0db8::1', 8080)            # 简化形式（省略 flowinfo 和 scope_id）

# AF_UNIX 地址示例
address_unix = '/tmp/mysocket.sock'              # Unix 域套接字路径

IP地址​

IP地址是计算机在网络中的唯一标识，由32位二进制数组成，分为IPv4和IPv6两种。

IPv4地址通常被写作点分十进制的形式，即四个字节被分开用十进制写出，中间用点分隔。即由4个字节组成，每个字节用点分隔，例如192.168.1.1。但实际上它可被写作任何表示一个32位整数值的形式?

IPv6地址通常被写作冒号分隔的形式，即由8个16位十六进制数组成，每个数用冒号分隔，例如2001:0db8:85a3:0000:0000:8a2e:0370:7334。

info

原则上每个设备只要连上公网，就必须有一个IP地址，这个IP地址是唯一的，不能重复。

但是目前世界上IP地址已经不够用了，所以如果需要遵循这个原则，则需使用IPv6地址。

如果不遵循这个原则可以在你的家里设置一个路由器，这个路由器可以上网，家里所有的设备把想和公网交换的数据发给这个路由器，然后这个路由再转发给公网。这个方案被称之为NAT(Network Address Translation)

特殊的IPV4地址​

这部分地址主要用于内部网络，比如你的家庭网络，公司网络，学校网络等。

名字地址范围地址数量有类别的描述最大的CIDR地址块
24位块10.0.0.0–10.255.255.25516,777,216一个A类10.0.0.0/8
20位块172.16.0.0–172.31.255.2551,048,576连续的16个B类172.16.0.0/12
16位块192.168.0.0–192.168.255.25565,536连续的256个C类192.168.0.0/16

回环地址: 127.0.0.1 ~ 127.255.255.255 用于测试本机网络。其中最常见的地址是127.0.0.1，其他地址在技术上完全等效，只是没有人使用罢了。

套接字类型​

类型常量说明
TCP流SOCK_STREAM面向连接的可靠传输，用于 TCP
UDP数据报SOCK_DGRAM无连接不可靠传输，用于 UDP
原始套接字SOCK_RAW原始网络协议访问?
warning

使用原始套接字需要管理员权限，且可以绕过操作系统的网络安全机制，可能被恶意软件利用。在生产环境中应谨慎使用。

工作流程​

类型工作流程
TCP服务器创建Socket → 绑定地址(bind) → 监听(listen) → 接受连接(accept) → 收发数据(send/recv) → 关闭Socket(close)
TCP客户端创建Socket → 连接服务器(connect) → 收发数据(send/recv) → 关闭Socket(close)
UDP服务器创建Socket → 绑定地址(bind) → 收发数据(recvfrom/sendto) → 关闭Socket(close)
UDP客户端创建Socket → (可选绑定) → 收发数据(sendto/recvfrom) → 关闭Socket(close)
原始套接字创建Socket → 绑定地址(bind) → 设置选项(setsockopt) → 收发原始数据包 → 关闭Socket(close)
tip

TCP 和 UDP 都是全双工通信，可以同时发送和接收数据。

  • TCP 的 send/recv：建立连接后可以使用 send() 发送数据，使用 recv() 接收数据，两个操作可以交替或并发进行
  • UDP 的 recvfrom/sendto：使用 recvfrom() 接收数据（返回数据和发送方地址），使用 sendto() 发送数据（需要指定目标地址）。UDP 无连接，每次发送都需指定地址

创建套接字​

socket.socket()​

创建套接字对象的基本方法：

import socket

# 创建 TCP 套接字（IPv4）
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 创建 UDP 套接字（IPv4）
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 使用默认参数（默认是 AF_INET, SOCK_STREAM）
sock = socket.socket()
tip

大多数情况下，使用 socket.socket() 创建 TCP 套接字就足够了，默认参数会自动选择最常用的配置。

套接字选项​

setsockopt()​

设置套接字选项：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 允许地址重用（避免 Address already in use 错误）
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 设置接收缓冲区大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8192)

# 设置发送缓冲区大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 8192)

# 绑定地址前设置选项
sock.bind(('localhost', 8080))

getsockopt()​

获取套接字选项的值：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 获取接收缓冲区大小
rcvbuf = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
print(f'接收缓冲区大小: {rcvbuf}')

# 检查 SO_REUSEADDR 选项
reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
print(f'地址重用: {bool(reuse)}')

setblocking()​

设置套接字为阻塞或非阻塞模式：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 设置为非阻塞模式
sock.setblocking(False)

# 设置为阻塞模式（默认）
sock.setblocking(True)

超时处理​

settimeout()​

设置套接字操作的超时时间（秒）：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 设置 5 秒超时
sock.settimeout(5.0)

try:
    sock.connect(('www.example.com', 80))
    data = sock.recv(1024)
except socket.timeout:
    print('操作超时')
except OSError as e:
    print(f'错误: {e}')

gettimeout()​

获取当前超时设置：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 获取超时设置（None 表示无超时）
timeout = sock.gettimeout()
print(f'超时设置: {timeout}')
info

超时设置影响以下操作：connect(), recv(), send(), accept()。如果设置为 None（默认），操作会无限期阻塞。

地址解析函数​

gethostname() 获取本地主机的标准主机名。 gethostbyname() 将主机名转换为 IPv4 地址。 gethostbyaddr() 将 IP 地址转换为主机名（反向解析）。 getservbyname() 根据服务名和协议获取端口号。 getservbyport() 根据端口号和协议获取服务名。

import socket

hostname = socket.gethostname()
print(hostname) # Allen

local_ip = socket.gethostbyname('localhost')
print(f'本地 IP: {local_ip}') # 127.0.0.1

# 解析主机名
ip = socket.gethostbyname('www.baidu.com')
print(f'IP 地址: {ip}') #183.2.172.177

hostname, aliases, addresses = socket.gethostbyname_ex('www.baidu.com')
print(f'主机名: {hostname}') # www.a.shifen.com
print(f'别名: {aliases}') # ['www.baidu.com']
print(f'IP 地址: {addresses}') #  ['183.2.172.177', '183.2.172.17']

gethostbyaddr()​

将 IP 地址转换为主机名（反向解析）：

import socket

try:
    hostname, aliases, addresses = socket.gethostbyaddr('93.184.216.34')
    print(f'主机名: {hostname}')
    print(f'别名: {aliases}')
    print(f'IP 地址: {addresses}')
except socket.herror:
    print('无法解析地址')

getservbyname()​

根据服务名和协议获取端口号：

import socket

# 获取 HTTP 服务的端口号
http_port = socket.getservbyname('http', 'tcp')
print(f'HTTP 端口: {http_port}')

# 获取 FTP 服务的端口号
ftp_port = socket.getservbyname('ftp', 'tcp')
print(f'FTP 端口: {ftp_port}')

# UDP 服务
dns_port = socket.getservbyname('domain', 'udp')
print(f'DNS 端口: {dns_port}')

服务器端操作​

bind()​

将套接字绑定到特定的网络地址：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定到本地地址和端口
sock.bind(('localhost', 8080))

# 绑定到所有网络接口
sock.bind(('', 8080))

# 绑定到 Unix 域套接字
unix_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
unix_sock.bind('/tmp/mysocket.sock')
warning

如果端口已被占用，bind() 会抛出 OSError: [Errno 98] Address already in use。可以使用 SO_REUSEADDR 选项来避免这个问题。

listen()、accept()​

listen() 使套接字进入监听状态，接受连接请求。

accept() 接受客户端连接，返回新的套接字对象和客户端地址。

accept() 会阻塞，直到有客户端连接。

import socket

sock = socket.socket()

sock.bind(('localhost', 8080))
sock.listen(5)

client_sock, client_addr = sock.accept()
print(f"""收到来自 {client_addr} 的连接""")

message = client_sock.recv(1024).decode('utf-8') or "空消息"
print(f"""消息为: {message}""")

客户端操作​

connect()、connect_ex()​

connect() 建立与服务器的连接。

connect_ex() 类似 connect()，但返回错误码而不是抛出异常。

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
    sock.connect(('localhost', 8080))
    print('连接成功')
except OSError as e:
    print(f'连接失败，错误码: {e.errno}')

err = sock.connect_ex(('localhost', 8080))
if err == 0:
    print('连接成功')
else:
    print(f'连接失败，错误码: {err}')

数据传输​

send()、sendall()、sendto()​

send() 发送数据，返回发送的字节数。 sendall() 发送所有数据，直到全部发送完成。 sendto() 发送数据到指定地址。

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('localhost', 8080))

# 发送数据（可能只发送部分数据）如果只发送了部分数据，需要继续发送剩余部分。
sent = sock.send(b'Hello, Server')
print(f'发送了 {sent} 字节')

# 发送所有数据，直到全部发送完成。确保所有数据都被发送
message = b'Hello, Server' * 100
sock.sendall(message)
print('所有数据已发送')

# 发送 UDP 数据到指定地址
sock.sendto(b'Hello, Server', ('localhost', 8080))

recv()、recvfrom()​

recv() 接收数据，返回接收到的数据。

recvfrom() 接收数据，返回接收到的数据和发送方地址。

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('localhost', 8080))

# 接收最多 1024 字节（1kb）
data = sock.recv(1024)
print(f'接收到 {len(data)} 字节: {data}')

data, addr = sock.recvfrom(1024)
print(f'从 {addr} 接收到: {data}')

关于最多接收字节数：

  • recv(bufsize) 中的 bufsize 参数指定单次调用最多接收的字节数
  • 理论上没有严格上限，但实际使用受以下因素限制：
    • 系统内存大小
    • 套接字接收缓冲区大小（SO_RCVBUF，通常默认 64KB 或更大）
    • 网络 MTU（最大传输单元，以太网通常为 1500 字节）
  • 实际接收的字节数可能少于 bufsize，取决于当前可用的数据量
  • 常用值：1024 (1KB)、2048 (2KB)、4096 (4KB)、8192 (8KB) 等（通常使用 2 的幂次）
tip

recv() 参数只是限制单次读取的最大字节数，不限制可以多次调用 recv() 接收的总数据量。如果需要接收固定大小的数据，需要在循环中多次调用直到接收完成。

关闭套接字​

close()​

关闭套接字，释放资源：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('www.example.com', 80))

# 关闭套接字
sock.close()

shutdown()​

关闭连接的一部分或全部：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('www.example.com', 80))

# SHUT_RD: 关闭接收
# SHUT_WR: 关闭发送
# SHUT_RDWR: 关闭接收和发送
sock.shutdown(socket.SHUT_RDWR)
sock.close()

getsockname()​

获取套接字自己的地址：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('localhost', 8080))

# 获取绑定后的地址
addr, port = sock.getsockname()
print(f'本地地址: {addr}:{port}')

getpeername()​

获取已连接套接字的远程地址：

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('www.example.com', 80))

# 获取远程地址
addr, port = sock.getpeername()
print(f'远程地址: {addr}:{port}')

示例：多线程TCP客户端和服务端​

使用标准库的 threading 模块实现TCP客户端和服务端，支持同时收发数据：

主线程负责监听客户端连接，每有一个客户端连接，就创建一个线程(handle_client)来处理该链接的收发数据。

该线程(handle_client)又包含两个线程(send_data, recv_data)来处理该链接的收发数据。

tcp_server.py
import socket
import threading

def handle_client(client_sock, client_addr):
    """处理客户端连接的线程函数"""
    print(f'客户端 {client_addr} 已连接')
    
    # 创建发送数据的线程
    def send_data():
        try:
            while True:
                message = input(f'向 {client_addr} 发送: ')
                if not message:
                    break
                client_sock.sendall(message.encode('utf-8'))
        except (ConnectionResetError, BrokenPipeError):
            print(f'客户端 {client_addr} 断开连接')
    
    # 创建接收数据的线程
    def recv_data():
        try:
            while True:
                data = client_sock.recv(1024)
                if not data:
                    break
                print(f'收到来自 {client_addr} 的消息: {data.decode("utf-8")}')
        except (ConnectionResetError, BrokenPipeError):
            print(f'客户端 {client_addr} 断开连接')
        finally:
            client_sock.close()
    
    # 启动发送和接收线程
    send_thread = threading.Thread(target=send_data, daemon=True)
    recv_thread = threading.Thread(target=recv_data, daemon=True)
    
    send_thread.start()
    recv_thread.start()
    
    # 等待接收线程结束（客户端断开连接）
    recv_thread.join()
    print(f'客户端 {client_addr} 连接已关闭')

def main():
    # 创建TCP套接字
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 允许地址重用（避免 Address already in use 错误）
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_sock.bind(('localhost', 8888))
    server_sock.listen(5)
    
    print('服务器启动，监听 0.0.0.0:8888')
    print('等待客户端连接...')
    
    try:
        while True:
            # 接受客户端连接
            client_sock, client_addr = server_sock.accept()
            
            # 为每个客户端创建新线程
            client_thread = threading.Thread(
                target=handle_client,
                args=(client_sock, client_addr),
                daemon=True
            )
            client_thread.start()
    except KeyboardInterrupt:
        print('\n服务器关闭')
    finally:
        server_sock.close()

if __name__ == '__main__':
    main()