Python socket基础

socket 模块实现底层网络通信，支持 TCP 和 UDP 协议。

TCP 客户端

import socket

# 创建 TCP socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 连接服务器
client.connect(('localhost', 8080))

# 发送数据
client.send(b'Hello Server')

# 接收数据
data = client.recv(1024)
print(f"收到: {data.decode()}")

# 关闭连接
client.close()

TCP 服务器

import socket

# 创建 TCP socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

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

# 监听连接
server.listen(5)
print("服务器启动，等待连接...")

while True:
    # 接受连接
    client_socket, address = server.accept()
    print(f"客户端连接: {address}")

    # 接收数据
    data = client_socket.recv(1024)
    print(f"收到: {data.decode()}")

    # 发送响应
    client_socket.send(b'Hello Client')

    # 关闭客户端连接
    client_socket.close()

UDP 客户端

import socket

# 创建 UDP socket（SOCK_DGRAM）
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# UDP 不需要连接，直接发送
client.sendto(b'Hello UDP', ('localhost', 8081))

# 接收数据
data, addr = client.recvfrom(1024)
print(f"收到: {data.decode()} 来自 {addr}")

client.close()

UDP 服务器

import socket

# 创建 UDP socket
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 绑定地址
server.bind(('localhost', 8081))
print("UDP 服务器启动...")

while True:
    # 接收数据
    data, addr = server.recvfrom(1024)
    print(f"收到: {data.decode()} 来自 {addr}")

    # 发送响应
    server.sendto(b'Hello UDP Client', addr)

socket 类型

socket 方法

设置超时

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.settimeout(5)  # 5秒超时

try:
    client.connect(('localhost', 8080))
except socket.timeout:
    print("连接超时")

client.close()

多客户端处理

import socket
import threading

def handle_client(client_socket, address):
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            client_socket.send(data)  # 回显
    finally:
        client_socket.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 8080))
server.listen(5)

while True:
    client_socket, address = server.accept()
    thread = threading.Thread(target=handle_client, args=(client_socket, address))
    thread.start()

文件传输

import socket

def send_file(host, port, filename):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect((host, port))

    # 发送文件名
    client.send(filename.encode())

    # 发送文件内容
    with open(filename, 'rb') as f:
        while True:
            data = f.read(1024)
            if not data:
                break
            client.send(data)

    client.close()

def receive_file(server_socket, save_dir):
    client_socket, addr = server_socket.accept()

    # 接收文件名
    filename = client_socket.recv(1024).decode()

    # 接收文件内容
    with open(f"{save_dir}/{filename}", 'wb') as f:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            f.write(data)

    client_socket.close()

端口和地址信息

import socket

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

# 本地地址
print(client.getsockname())  # 本地 IP 和端口

# 远端地址
print(client.getpeername())  # 服务器 IP 和端口

client.close()

错误处理

import socket

try:
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(('localhost', 8080))
    client.send(b'data')
except socket.error as e:
    print(f"Socket 错误: {e}")
finally:
    client.close()

要点总结

• socket(AF_INET, SOCK_STREAM) 创建 TCP socket
• socket(AF_INET, SOCK_DGRAM) 创建 UDP socket
• TCP：bind -> listen -> accept -> recv/send
• UDP：bind -> recvfrom/sendto（无连接）
• settimeout() 设置超时避免阻塞
• 多线程处理多个客户端连接
• getsockname() 获取本地地址
• getpeername() 获取远端地址
• socket 是底层网络编程的核心模块