一、库的简介：网络通信的基石

你有没有想过，当你用浏览器访问一个网站时，你的电脑和远方的服务器是如何建立连接、交换数据的？当你用一个手机 App 给朋友发消息时，信息是怎样穿越复杂的网络最终到达对方手机的？这一切的背后，都离不开一个叫作“套接字”（Socket）的核心技术。而 Python 的 socket 标准库，正是对这套底层网络接口的封装，它允许你直接创建 TCP 或 UDP 连接，发送和接收原始数据，实现任意自定义的网络协议。在实际生活中，socket 可以用于编写一个局域网聊天工具，让办公室的同事在不连外网的情况下交流；可以写一个简单的文件传输程序，代替 U 盘在两台电脑间复制大文件；也可以搭建一个自己的 HTTP 服务器（虽然性能比不上 Nginx，但胜在轻量和灵活）。可以说，socket 是所有网络应用的基石——从 Web 服务器、数据库客户端，到即时通讯软件、远程桌面，底层都离不开套接字编程。掌握 socket，你就能真正理解“网络通信”到底是怎么回事，并能自己动手实现各种网络工具。

二、安装 socket

socket 是 Python 的标准库模块，随 Python 解释器一起安装，无需任何 pip install。直接导入即可：

python

import socket

三、基本用法

下面以最常用的 TCP 协议为例，分 4 步展示一个客户端-服务器通信的完整流程。

第一步：创建 TCP 服务器端

服务器需要绑定一个地址和端口，监听来自客户端的连接。

python

import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('127.0.0.1', 8888))   # 绑定本地 8888 端口
server_socket.listen(5)                   # 开始监听，最大等待队列 5
print("服务器已启动，等待连接...")

client_socket, client_addr = server_socket.accept()  # 阻塞直到有客户端连接
print(f"客户端 {client_addr} 已连接")

data = client_socket.recv(1024)           # 接收最多 1024 字节
print(f"收到: {data.decode()}")
client_socket.send(b"Hello from server")  # 发送响应

client_socket.close()
server_socket.close()

第二步：创建 TCP 客户端

客户端主动连接服务器的 IP 和端口，发送数据并接收响应。

python

import socket

client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(('127.0.0.1', 8888))

client_socket.send(b"Hello from client")
response = client_socket.recv(1024)
print(f"服务器响应: {response.decode()}")

client_socket.close()

第三步：UDP 通信（无连接）

UDP 不需要建立连接，直接发送数据报到目标地址。

python

# UDP 服务器
udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_server.bind(('127.0.0.1', 9999))
data, addr = udp_server.recvfrom(1024)
print(f"来自 {addr} 的消息: {data.decode()}")
udp_server.sendto(b"ACK", addr)

# UDP 客户端
udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_client.sendto(b"Hello UDP", ('127.0.0.1', 9999))
response, _ = udp_client.recvfrom(1024)
print(response.decode())

第四步：异常处理与超时设置

网络通信充满不确定性，设置超时和捕获异常非常重要。

python

client_socket.settimeout(5.0)   # 设置 5 秒超时
try:
    data = client_socket.recv(1024)
except socket.timeout:
    print("接收超时")
except socket.error as e:
    print(f"Socket 错误: {e}")

四、高级用法

1. 非阻塞模式

将 socket 设置为非阻塞，recv / accept 不会等待，而是立即返回错误（BlockingIOError）。通常与 select 或 asyncio 配合实现单线程处理多个连接。

python

server_socket.setblocking(False)

2. 使用 selectors 实现 I/O 多路复用

selectors 模块（Python 3.4+）封装了底层的 select/epoll，可以高效监控多个 socket 的事件。

python

import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock, mask):
    conn, addr = sock.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    data = conn.recv(1024)
    if data:
        print(f"收到: {data.decode()}")
        conn.send(b"Echo: " + data)
    else:
        sel.unregister(conn)
        conn.close()

server = socket.socket()
server.bind(('127.0.0.1', 8888))
server.listen()
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

3. 发送和接收完整数据（处理粘包）

TCP 是流式协议，一次 send 可能分多次 recv 才能收完，需要自己定义消息边界（如先发送长度头）。

python

def send_all(sock, data):
    sock.sendall(len(data).to_bytes(4, 'big'))   # 先发送 4 字节长度
    sock.sendall(data)

def recv_all(sock):
    length_data = sock.recv(4)
    if not length_data:
        return None
    length = int.from_bytes(length_data, 'big')
    data = b''
    while len(data) < length:
        chunk = sock.recv(min(4096, length - len(data)))
        if not chunk:
            raise ConnectionError("连接断开")
        data += chunk
    return data

五、实际应用场景案例

场景一：局域网内传输文件（TCP）

在日常办公中，经常需要在两台电脑之间传输大文件。用 U 盘拷贝麻烦，网盘上传下载又慢。下面是一个简单的文件传输程序，接收端运行服务器，发送端连接并发送文件。

python

# 文件发送端（客户端）
import socket
import os

def send_file(host, port, file_path):
    if not os.path.isfile(file_path):
        print("文件不存在")
        return
    file_name = os.path.basename(file_path)
    file_size = os.path.getsize(file_path)

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    # 发送文件名（固定 256 字节）和文件大小（8 字节）
    sock.send(file_name.ljust(256).encode())
    sock.send(file_size.to_bytes(8, 'big'))

    with open(file_path, 'rb') as f:
        sent = 0
        while sent < file_size:
            chunk = f.read(8192)
            if not chunk:
                break
            sock.sendall(chunk)
            sent += len(chunk)
            print(f"进度: {sent*100//file_size}%", end='\r')
    print("\n发送完成")
    sock.close()

if __name__ == '__main__':
    send_file('192.168.1.100', 12345, './report.pdf')

python

# 文件接收端（服务器）
import socket
import os

def receive_file(port, save_dir='.'):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(('0.0.0.0', port))
    server.listen(1)
    print(f"等待文件传输，监听端口 {port}...")
    conn, addr = server.accept()
    print(f"来自 {addr} 的连接")

    # 接收文件名和大小
    raw_name = conn.recv(256).decode().strip('\x00')
    if not raw_name:
        return
    file_size = int.from_bytes(conn.recv(8), 'big')
    save_path = os.path.join(save_dir, raw_name)

    with open(save_path, 'wb') as f:
        received = 0
        while received < file_size:
            chunk = conn.recv(8192)
            if not chunk:
                break
            f.write(chunk)
            received += len(chunk)
            print(f"接收进度: {received*100//file_size}%", end='\r')
    print(f"\n文件保存为 {save_path}")
    conn.close()
    server.close()

if __name__ == '__main__':
    receive_file(12345)

场景二：简易 HTTP 服务器（单线程）

在开发前端页面时，需要一个简单的 HTTP 服务器来测试静态文件。用 socket 手写一个极简版，能返回 HTML 内容。

python

import socket

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

while True:
    client, addr = server.accept()
    request = client.recv(1024).decode()
    print(request[:100])  # 打印请求第一行

    # 构建 HTTP 响应
    html = """<html><body><h1>Hello from custom socket server</h1></body></html>"""
    response = f"""HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: {len(html)}\r\n\r\n{html}"""
    client.sendall(response.encode())
    client.close()

场景三：TCP 端口扫描器（安全测试）

管理员可以用它检查自己的服务器开放了哪些端口，防止危险端口暴露。

python

import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        if result == 0:
            return port
        sock.close()
    except:
        pass
    return None

def port_scan(host, start_port, end_port, max_workers=100):
    open_ports = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(scan_port, host, port) for port in range(start_port, end_port+1)]
        for future in futures:
            port = future.result()
            if port:
                open_ports.append(port)
    return open_ports

if __name__ == '__main__':
    host = '127.0.0.1'
    ports = port_scan(host, 1, 1024)
    print(f"开放端口: {ports}")

六、结尾互动

socket 是网络编程的基石，它让你深入理解 TCP/IP 协议栈的运作方式，摆脱对高层框架的盲目依赖。虽然现代开发中我们更多地使用 requests、aiohttp、FastAPI 等封装好的库，但当你遇到特殊协议或需要极致性能优化时，socket 依然是最终武器。从简单的聊天室、文件传输，到定制化的协议解析，socket 都能帮你实现。掌握它，你就能真正“驾驭”网络。

你有没有用 socket 实现过什么有趣的工具？比如一个网络五子棋对弈程序，或者一个远程执行命令的后门（仅供学习）？欢迎在评论区分享你的创意。不妨今天就动手写一个“网络版猜数字”游戏，邀请朋友一起玩，感受 Socket 的魅力！

# 网络 # python # 开发语言
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

cover

安装ubuntu1804操作系统

avatar openEuler 社区

使用 Python 转储LSASS凭证和提取使用

LSASS 是Windows操作系统非常重要的系统服务，因为它保存着每个登录用户的加密密码和令牌数据，所以也成了网络攻击系统的首要目标。一旦这些凭据被盗，攻击者就可以冒充用户、提升权限或在网络中不受控制地移动。本文用Python实现了转储LSASS凭证和提取使用的程序，验证了LSASS凭证转储怎样规避特权上下文、未挂钩的内存读取以及对文件写入的隐秘处理，实现有效且隐秘地自动提取 LSASS 凭证。

avatar openEuler 社区
cover

黑马程序员苍穹外卖--学习笔记（苍穹外卖万字总结—重点知识，面试常见问题）超全哦

avatar openEuler 社区