人工智能之编程进阶 Python高级
第八章 网络并发类模块
前言
本文针对于网络编程以及网络编程中的多线程以及异步编程和子进程进行详细的叙述，并对这几种方式进行了对比总结。
---# 🌐 一、网络编程（socket）
Python 使用
socket 模块进行底层网络通信，支持 TCP/UDP。
1. TCP 服务端（监听连接）
import socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 8888))
server.listen(5) # 最多排队 5 个连接
print("服务器启动，等待连接...")
while True:
client, addr = server.accept()
print(f"客户端 {addr} 已连接")
data = client.recv(1024).decode('utf-8')
print("收到:", data)
client.send(b"Hello from server!")
client.close()
2. TCP 客户端
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('localhost', 8888))
client.send(b"Hello server!")
response = client.recv(1024)
print("服务器回复:", response.decode())
client.close()
3. UDP 示例（无连接）
# 服务端
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('localhost', 9999))
data, addr = sock.recvfrom(1024)
sock.sendto(b"Got it!", addr)
# 客户端
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.sendto(b"Hi UDP!", ('localhost', 9999))
data, _ = sock.recvfrom(1024)
✅ 适用场景：自定义协议、学习网络原理 ❌ 不推荐：写 Web 服务（用 Flask/FastAPI 更高效）
🔁 二、多线程（threading）
用于 I/O 密集型任务（如文件读写、网络请求），但受 GIL 限制， 不能真正并行 CPU 计算。
1. 基本用法
import threading
import time
def worker(name):
print(f"线程 {name} 开始")
time.sleep(2)
print(f"线程 {name} 结束")
# 创建并启动线程
t1 = threading.Thread(target=worker, args=("A",))
t2 = threading.Thread(target=worker, args=("B",))
t1.start()
t2.start()
t1.join() # 等待线程结束
t2.join()
print("所有线程完成")
2. 线程安全（加锁）
lock = threading.Lock() counter = 0 def increment(): global counter for _ in range(100000): with lock: # 或 lock.acquire() / lock.release() counter += 1 t1 = threading.Thread(target=increment) t2 = threading.Thread(target=increment) t1.start(); t2.start() t1.join(); t2.join() print(counter) # 应为 200000
⚠️ 注意：
- Python 的 GIL（全局解释器锁） 使得同一时间只有一个线程执行 Python 字节码。
- 所以多线程适合 I/O 阻塞场景（如下载、数据库查询），不适合 CPU 密集型任务。
✅ 典型用途：同时处理多个客户端请求、并发爬虫（配合 requests）
⚡ 三、异步 I/O（asyncio）
Python 的
原生异步框架，使用
async/
await 实现单线程高并发，特别适合
大量 I/O 操作（如 HTTP 请求、数据库查询）。
1. 基本异步函数
import asyncio
async def say_hello(delay, name):
await asyncio.sleep(delay) # 异步等待（不阻塞其他任务）
print(f"Hello {name}!")
async def main():
# 并发运行多个协程
await asyncio.gather(
say_hello(1, "Alice"),
say_hello(2, "Bob"),
say_hello(0.5, "Charlie")
)
# 启动事件循环
asyncio.run(main())
2. 异步 HTTP 请求（需第三方库）
import aiohttp
import asyncio
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
urls = ["https://httpbin.org/delay/1"] * 5
tasks = [fetch(url) for url in urls]
results = await asyncio.gather(*tasks)
print("全部完成！")
asyncio.run(main())
✅ 优势：
- 单线程高并发（万级连接）
- 资源占用低
- 代码逻辑清晰（避免回调地狱）
❌ 限制：
- 所有 I/O 操作必须是 异步兼容的（不能直接用
time.sleep()、requests.get()）- 不适合 CPU 密集型任务（会阻塞事件循环）
📌 适用场景：高性能 Web 服务（FastAPI、aiohttp）、实时聊天、爬虫
🧒 四、子进程（subprocess）
用于 启动和控制外部程序（如 shell 命令、其他语言脚本），可绕过 GIL，实现真正的 CPU 并行。
1. 简单运行命令
import subprocess # 运行命令并获取输出 result = subprocess.run(["ls", "-l"], capture_output=True, text=True) print(result.stdout)
2. 启动长期运行的进程
proc = subprocess.Popen(
["ping", "baidu.com"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 读取输出（非阻塞需配合 select 或线程）
try:
stdout, stderr = proc.communicate(timeout=10)
print(stdout)
except subprocess.TimeoutExpired:
proc.kill()
print("超时，已终止")
3. 与多进程结合（
multiprocessing）
from multiprocessing import Pool import os def cpu_task(n): return sum(i*i for i in range(n)) # 纯 CPU 计算 if __name__ == '__main__': with Pool() as pool: # 自动使用 CPU 核心数 results = pool.map(cpu_task, [1000000] * 4) print(results)
✅ 关键点：
subprocess适合调用 外部程序multiprocessing适合 Python 内部 CPU 并行- 两者都能绕过 GIL，实现真正并行
📌 典型用途：
- 调用 ffmpeg 处理视频
- 执行 shell 脚本
- 并行处理图像/数据（用
multiprocessing）
🔄 五、进程 vs 线程：本质区别与选型建议
这是很多初学者容易混淆的核心问题。下面从多个维度对比：
| 对比项 | 线程（Thread） | 进程（Process） |
| ----------------------- | ------------------------------------ | ------------------------------------------ |
| 定义 | 同一进程内的执行单元 | 操作系统分配资源的基本单位 |
| 内存空间 | 共享进程内存（变量、堆） | 独立内存空间（不共享） |
| 创建开销 | 小（轻量） | 大（需复制内存、资源） |
| 通信方式 | 直接读写共享变量（需加锁） | 需通过 IPC（如 Queue、Pipe、共享内存） |
| 稳定性 | 一个线程崩溃 → 整个进程崩溃 | 一个进程崩溃 → 不影响其他进程 |
| GIL 影响 | 受 GIL 限制，Python 中无法并行 CPU | 绕过 GIL，真正并行 CPU |
| 适用任务 | I/O 密集型（网络、磁盘） | CPU 密集型（计算、图像处理） |
|
Python 模块 |
threading |
multiprocessing /
subprocess |
📌 一句话总结：
- 线程 = 共享内存 + 轻量 + 适合等 I/O
- 进程 = 独立内存 + 重量 + 适合干 CPU 活
💡 实际选型建议：
-
爬取 100 个网页？ → 用
threading或asyncio（都在等网络响应） -
计算 100 万个数的平方和？ → 用
multiprocessing（真正并行计算） -
调用外部程序（如 ffmpeg）？ → 用
subprocess（天然隔离、安全）
⚠️ 注意： 在 Python 中，由于 GIL（全局解释器锁） 的存在， 多线程无法提升 CPU 密集型任务的性能。这是选择进程而非线程的关键原因。
🔍 六、四大方式对比总结
| 方式 | 是否并行 | 适用任务类型 | 是否绕过 GIL | 典型场景 |
| ------------------------------------------ | -------------- | -------------- | -------------- | ------------------------------ |
|
多线程 (
threading) | ❌（伪并行） | I/O 密集型 | ❌ | 爬虫、文件读写、多客户端 |
|
异步 I/O (
asyncio) | ❌（单线程） | I/O 密集型 | ❌ | 高并发 Web 服务、实时通信 |
|
子进程 (
subprocess) | ✅ | 调用外部程序 | ✅ | 执行 shell、调用 C/Java 程序 |
|
多进程 (
multiprocessing) | ✅ | CPU 密集型 | ✅ | 数据计算、图像处理、科学计算 |
💡 一句话选型：
- 等网络/磁盘？ → 用
threading或asyncio- 算数学/处理数据？ → 用
multiprocessing- 跑外部命令？ → 用
subprocess
✅ 七、最佳实践建议
-
Web 服务优先选
asyncio（如 FastAPI + async/await），性能远超多线程。 -
不要在 asyncio 中混用阻塞代码（如
time.sleep()、requests），会卡住整个事件循环。 -
CPU 密集任务别用 threading，改用
multiprocessing。 - 子进程记得处理超时和异常，避免僵尸进程。
-
线程间共享数据务必加锁（
Lock、Queue）。 -
开发阶段可用
concurrent.futures统一接口：
``
python from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor ``
资料关注
公众号：咚咚王
《Python编程：从入门到实践》 《利用Python进行数据分析》 《算法导论中文第三版》 《概率论与数理统计（第四版） (盛骤) 》 《程序员的数学》 《线性代数应该这样学第3版》 《微积分和数学分析引论》 《（西瓜书）周志华-机器学习》 《TensorFlow机器学习实战指南》 《Sklearn与TensorFlow机器学习实用指南》 《模式识别（第四版）》 《深度学习 deep learning》伊恩·古德费洛著 花书 《Python深度学习第二版(中文版)【纯文本】 (登封大数据 (Francois Choliet)) (Z-Library)》 《深入浅出神经网络与深度学习+(迈克尔·尼尔森（Michael+Nielsen） 》 《自然语言处理综论 第2版》 《Natural-Language-Processing-with-PyTorch》 《计算机视觉-算法与应用(中文版)》 《Learning OpenCV 4》 《AIGC：智能创作时代》杜雨+&+张孜铭 《AIGC原理与实践：零基础学大语言模型、扩散模型和多模态模型》 《从零构建大语言模型（中文版）》 《实战AI大模型》 《AI 3.0》
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70048224/viewspace-3102500/ ，如需转载，请注明出处，否则将追究法律责任。
