Python模块详解（一）—— socket 和 threading 模块

MIXLLRED

500人浏览 · 2026-03-15 21:12:32

MIXLLRED · 2026-03-15 21:12:32 发布

（一）Python socket 模块详解

socket 模块提供了对 Berkeley sockets API 的访问，是实现网络通信的基础。通过它，可以创建客户端和服务器，进行 TCP 或 UDP 数据传输。

1. 核心概念

（1）地址族（Address Family）：指定通信使用的协议。

AF_INET：IPv4（最常用）

AF_INET6：IPv6

AF_UNIX：Unix 域套接字（本地进程间通信）

（2）套接字类型（Socket Type）：

SOCK_STREAM：面向连接的可靠字节流（TCP）

SOCK_DGRAM：无连接的数据报（UDP）

SOCK_RAW：原始套接字（通常需要管理员权限）

（3）协议（Protocol）：通常设为 0，让系统自动选择。

2. 主要函数与方法

（1）创建 socket

import socket
s = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0)

family：默认为 AF_INET，type：默认为 SOCK_STREAM。

（2）服务器端常用方法

bind(address)：将套接字绑定到指定地址（IP 和端口）。
address 格式：(host, port)，例如 ('127.0.0.1', 8888)。

listen([backlog])：开始监听连接请求。backlog 指定最大挂起连接数。

accept()：阻塞等待客户端连接，返回 (conn, address)，其中 conn 是新的套接字对象，用于与客户端通信。

（3）客户端常用方法

connect(address)：连接到远程服务器。

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

（4）数据收发

send(bytes)：发送数据，返回实际发送的字节数（可能小于数据长度）。
可使用 sendall() 确保全部发送。

recv(bufsize[, flags])：接收数据，最多 bufsize 字节，返回接收到的字节串。

sendto(bytes, address)：UDP 发送数据到指定地址。

recvfrom(bufsize)：UDP 接收数据，返回 (data, address)。

（5）控制选项

setsockopt(level, optname, value) / getsockopt(level, optname)：设置/获取套接字选项。例如 socket.SO_REUSEADDR 允许重用地址。

setblocking(flag)：设置阻塞模式。flag=False 时套接字为非阻塞，若操作无法立即完成则抛出 BlockingIOError。

settimeout(value)：设置超时时间（秒）。超时后操作抛出 timeout 异常。

（6）关闭套接字

close()：释放资源。

（7）辅助函数

gethostname()：获取本地主机名。

gethostbyname(hostname)：将主机名解析为 IPv4 地址。

gethostbyaddr(ip_address)：反向解析。

getaddrinfo(host, port, family=0, type=0, proto=0, flags=0)：获取地址信息列表，是编写协议无关代码的推荐方式。

3. 使用示例

（1）TCP 服务器

具体示例如下：

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 允许端口重用
server.bind(('0.0.0.0', 8888))
server.listen(5)
print("Server listening on port 8888")

while True:
    conn, addr = server.accept()
    print(f"Connected by {addr}")
    data = conn.recv(1024)
    if data:
        conn.sendall(b"Received: " + data)
    conn.close()

（2）TCP 客户端

具体示例如下：

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('127.0.0.1', 8888))
client.sendall(b"Hello, server!")
response = client.recv(1024)
print("Received:", response)
client.close()

（3）UDP 服务器

import socket

udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_server.bind(('0.0.0.0', 8889))
while True:
    data, addr = udp_server.recvfrom(1024)
    print(f"Message from {addr}: {data}")
    udp_server.sendto(b"ACK", addr)

4. 注意事项

异常处理：网络操作可能引发 socket.error、timeout 等异常，应使用 try/except 捕获。
阻塞与非阻塞：默认情况下，accept()、recv() 等会阻塞线程，可使用 setblocking(False) 或 settimeout() 改变行为。
字节与字符串：网络传输的是字节数据，发送前需将字符串编码（如 encode()），接收后解码（decode()）。

（二）Python threading 模块详解

threading 模块提供了高级的多线程编程接口，允许在单个进程中并发执行多个任务。

1. 核心概念

（1）线程（Thread）：轻量级进程，共享进程的内存空间。

（2）全局解释器锁（GIL）：CPython 中，GIL 使得同一时刻只有一个线程执行 Python 字节码，因此多线程并不能充分利用多核 CPU 进行并行计算，但在 I/O 密集型任务中依然有效。

（3）线程安全：多个线程同时访问共享数据时，需要同步机制避免数据竞争。

2. 主要类与函数

（1）Thread 类

创建线程：

 t = threading.Thread(target=func, args=(arg1,), kwargs={}, name='Thread-1')

其中：

target：线程执行的函数。

args：位置参数元组。

kwargs：关键字参数字典。

name：线程名称。

启动线程：

t.start()

等待线程结束：

t.join(timeout)

阻塞直到线程终止。

其他属性：t.ident（线程标识符）、t.is_alive()（是否存活）、t.daemon（是否为守护线程）。

（2）线程同步原语

Lock：互斥锁。

acquire(blocking=True, timeout=-1)：获取锁，可阻塞。
release()：释放锁。

示例：

lock = threading.Lock()
lock.acquire()
# 临界区
lock.release()

更推荐使用 with lock: 上下文管理器。

RLock：可重入锁，同一线程可多次获取，避免死锁。

Semaphore：信号量，允许一定数量的线程同时访问资源。

Event：事件，用于线程间通信。set()、wait()、clear()。

Condition：条件变量，可用于生产者-消费者模式。

线程间通信

queue.Queue：线程安全的队列，常用于数据交换。

put(item)：入队。
get()：出队，若队列空则阻塞。

（3）模块级函数

threading.active_count()：当前存活的线程数。

threading.current_thread()：返回当前线程对象。

threading.enumerate()：返回所有存活线程的列表。

threading.main_thread()：返回主线程对象。

3. 使用示例

（1）创建并启动线程

import threading

import time

def worker(name):
    print(f"Thread {name} starting")
    time.sleep(2)
    print(f"Thread {name} finished")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()
print("All threads done")

（2）使用锁避免竞争条件

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 正确输出 200000

（3）使用 Queue 进行线程间通信（生产者-消费者）

import threading
import queue
import time

def producer(q):
    for i in range(5):
        item = f"item-{i}"
        q.put(item)
        print(f"Produced {item}")
        time.sleep(1)
    q.put(None)  # 结束信号

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumed {item}")
        time.sleep(1.5)

q = queue.Queue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start()
t2.start()
t1.join()
t2.join()

4. 注意事项

守护线程（Daemon Thread）：设置 t.daemon = True（或在构造时指定），主线程退出时守护线程会自动终止，不会等待。
避免死锁：多个锁嵌套使用时，确保获取顺序一致，或使用 RLock。
线程安全：不要共享可变对象而不加锁，使用 Queue 或 threading 同步原语。
GIL 的影响：对于 CPU 密集型任务，多线程可能比单线程还慢，可考虑 multiprocessing 模块。

（三）socket 和 threading 结合使用

socket 和 threading 常结合使用，例如在多线程服务器中，每个客户端连接分配一个线程处理。典型模式：

import socket
import threading

def handle_client(conn, addr):
    with conn:
        print(f"Thread {threading.current_thread().name} handling {addr}")
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)

server = socket.socket()
server.bind(('0.0.0.0', 8888))
server.listen(5)
while True:
    conn, addr = server.accept()
    t = threading.Thread(target=handle_client, args=(conn, addr))
    t.start()

这样每个连接独立处理，主线程继续等待新连接。