Python多进程与多线程适用场景案例分析
<div id="navCategory"><h5 class="catalogue">目录</h5><ul class="first_class_ul"><li><a href="#_label0">一、多线程（threading）：主打“IO密集型”场景</a></li><ul class="second_class_ul"><li><a href="#_lab2_0_0">1. 核心适用场景（附案例）</a></li><ul class="third_class_ul"><li><a href="#_label3_0_0_0">（1）网络相关任务（最典型）</a></li><li><a href="#_label3_0_0_1">（2）文件/数据库IO任务</a></li><li><a href="#_label3_0_0_2">（3）需频繁数据共享的轻量任务</a></li></ul><li><a href="#_lab2_0_1">2. 多线程避坑场景</a></li><ul class="third_class_ul"></ul></ul><li><a href="#_label1">二、多进程（multiprocessing）：主打“CPU密集型”场景</a></li><ul class="second_class_ul"><li><a href="#_lab2_1_2">1. 核心适用场景（附案例）</a></li><ul class="third_class_ul"><li><a href="#_label3_1_2_3">（1）数据计算类任务（最典型）</a></li><li><a href="#_label3_1_2_4">（2）批量独立任务处理</a></li><li><a href="#_label3_1_2_5">（3）高稳定性要求的任务</a></li></ul><li><a href="#_lab2_1_3">2. 多进程避坑场景</a></li><ul class="third_class_ul"></ul></ul><li><a href="#_label2">三、选型速查表（新手直接套用）</a></li><ul class="second_class_ul"></ul><li><a href="#_label3">四、特殊场景：混合使用（线程+进程）</a></li><ul class="second_class_ul"></ul></ul></div><p>你想明确多进程和多线程各自的适用场景，核心是要结合任务类型、资源需求、数据共享等维度来判断——简单来说，<strong>IO密集型任务优先用多线程，CPU密集型任务必须用多进程</strong>，下面拆解具体场景和选型逻辑，附实战案例帮你落地。</p><p class="maodian"><a name="_label0"></a></p><h2>一、多线程（threading）：主打“IO密集型”场景</h2>
<p>多线程的核心优势是“轻量、切换快、数据共享方便”，但受GIL限制无法利用多核，因此只适合“等待时间远大于计算时间”的任务（IO等待时线程释放GIL，其他线程可执行）。</p>
<p class="maodian"><a name="_lab2_0_0"></a></p><p class="maodian"><a name="_lab2_1_2"></a></p><h3>1. 核心适用场景（附案例）</h3>
<p class="maodian"><a name="_label3_0_0_0"></a></p><h4>（1）网络相关任务（最典型）</h4>
<ul><li>爬虫（批量爬取网页、接口数据）、API调用、消息队列消费、网络监控；</li><li>案例：多线程爬取10个网页（IO等待占90%）</li></ul>
<div class="jb51code"><pre class="brush:py;">import threading
import time
import requests
def crawl(url):
print(f"线程{threading.current_thread().name}：请求{url}")
resp = requests.get(url) 网络IO等待（释放GIL）
print(f"{url}：状态码{resp.status_code}")
if __name__ == "__main__":
start = time.time()
urls =
threads =
print(f"总耗时：{time.time()-start:.2f}秒") 约2秒（并发），单线程约10秒</pre></div>
<p class="maodian"><a name="_label3_0_0_1"></a></p><h4>（2）文件/数据库IO任务</h4>
<ul><li>批量读写本地文件、数据库查询/插入（如MySQL/Redis操作）、日志采集；</li><li>特点：这类任务大部分时间在“等磁盘/数据库响应”，线程切换开销可忽略；</li><li>案例：多线程读取10个大文本文件</li></ul>
<div class="jb51code"><pre class="brush:py;">from concurrent.futures import ThreadPoolExecutor
import os
def read_file(filepath):
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
return f"{filepath}：字符数{len(content)}"
if __name__ == "__main__":
files = [:10]
with ThreadPoolExecutor(max_workers=5) as executor:
results = executor.map(read_file, files)
for res in results:
print(res)</pre></div>
<p class="maodian"><a name="_label3_0_0_2"></a></p><h4>（3）需频繁数据共享的轻量任务</h4>
<ul><li>实时统计（如统计程序运行状态、实时计数）、GUI程序后台任务（如界面刷新+数据加载）；</li><li>特点：线程共享进程内存，无需复杂通信，加锁即可保证数据安全；</li><li>案例：多线程实时计数（共享变量）</li></ul>
<div class="jb51code"><pre class="brush:py;">import threading
import time
count = 0
lock = threading.Lock()
def add_count():
global count
for _ in range(10000):
with lock: 加锁避免数据竞争
count += 1
if __name__ == "__main__":
t1 = threading.Thread(target=add_count)
t2 = threading.Thread(target=add_count)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"最终计数：{count}") 输出20000（准确）</pre></div>
<p class="maodian"><a name="_lab2_0_1"></a></p><h3>2. 多线程避坑场景</h3>
<ul><li>❌ 绝对不要用多线程处理CPU密集任务（如矩阵运算、数据加密）：GIL会让多线程变成“伪并发”，效率甚至低于单线程；</li><li>❌ 不要创建过多线程（如上千个）：线程切换开销累积，会导致程序卡顿。</li></ul>
<p class="maodian"><a name="_label1"></a></p><h2>二、多进程（multiprocessing）：主打“CPU密集型”场景</h2>
<p>多进程的核心优势是“突破GIL限制、利用多核CPU、进程隔离更稳定”，但启动/通信开销大，适合“计算时间占主导”的任务。</p>
<h3>1. 核心适用场景（附案例）</h3>
<p class="maodian"><a name="_label3_1_2_3"></a></p><h4>（1）数据计算类任务（最典型）</h4>
<ul><li>矩阵运算、数值模拟、数据挖掘（如批量特征计算）、密码破解、AI模型训练（单机多进程）；</li><li>案例：多进程计算10个大矩阵的乘法</li></ul>
<div class="jb51code"><pre class="brush:py;">from concurrent.futures import ProcessPoolExecutor
import time
import numpy as np
def matrix_calc(matrix):
return np.dot(matrix, matrix) 纯CPU计算
if __name__ == "__main__":
start = time.time()
matrices =
with ProcessPoolExecutor() as executor:
executor.map(matrix_calc, matrices)
print(f"总耗时：{time.time()-start:.2f}秒") 4核CPU约3秒，单进程约10秒</pre></div>
<p class="maodian"><a name="_label3_1_2_4"></a></p><h4>（2）批量独立任务处理</h4>
<ul><li>图片/视频处理（如批量压缩图片、视频转码）、文件格式转换（如CSV转Excel）、批量数据分析；</li><li>特点：任务间无依赖、可独立执行，进程隔离避免一个任务崩溃影响全局；</li><li>案例：多进程批量压缩图片</li></ul>
<div class="jb51code"><pre class="brush:py;">import multiprocessing
from PIL import Image
import os
def compress_img(filepath):
img = Image.open(filepath)
img.save(f"compressed_{filepath}", quality=50) 压缩图片（CPU密集）
return f"{filepath}压缩完成"
if __name__ == "__main__":
imgs = [:5]
pool = multiprocessing.Pool(processes=4)
results = pool.map(compress_img, imgs)
pool.close()
pool.join()
for res in results:
print(res)</pre></div>
<p class="maodian"><a name="_label3_1_2_5"></a></p><h4>（3）高稳定性要求的任务</h4>
<ul><li>批量执行第三方脚本/工具、不稳定的任务（如调用易崩溃的接口）；</li><li>特点：单个进程崩溃不会导致整个程序挂掉，容错性更高。</li></ul>
<p class="maodian"><a name="_lab2_1_3"></a></p><h3>2. 多进程避坑场景</h3>
<ul><li>❌ 不要用多进程处理高频小任务（如每秒执行一次的定时检测）：进程启动/通信开销会抵消并行优势；</li><li>❌ 不要用多进程处理需频繁数据共享的任务：进程间通信（Queue/Pipe）开销大，效率远低于线程。</li></ul>
<p class="maodian"><a name="_label2"></a></p><h2>三、选型速查表（新手直接套用）</h2>
<table><thead><tr><th>任务特征</th><th>优先选择</th><th>核心原因</th></tr></thead><tbody><tr><td>CPU密集（计算为主）</td><td>多进程</td><td>突破GIL，利用多核并行计算</td></tr><tr><td>IO密集（等待为主）</td><td>多线程</td><td>轻量并发，IO等待时释放GIL</td></tr><tr><td>需频繁共享数据</td><td>多线程</td><td>共享内存，通信成本低</td></tr><tr><td>任务独立、无数据共享</td><td>多进程</td><td>进程隔离，稳定性高</td></tr><tr><td>任务数量多、启动频繁</td><td>多线程</td><td>线程启动快，切换开销低</td></tr><tr><td>单个任务易崩溃</td><td>多进程</td><td>进程隔离，不影响其他任务</td></tr><tr><td>单机最大化算力</td><td>多进程</td><td>充分利用多核CPU</td></tr></tbody></table>
<p class="maodian"><a name="_label3"></a></p><h2>四、特殊场景：混合使用（线程+进程）</h2>
<p>如果任务是“IO+CPU混合”（如爬取数据后立即计算），可结合二者优势：</p>
<p>示例：多进程（处理CPU计算）+ 多线程（处理IO爬取）</p>
<div class="jb51code"><pre class="brush:py;">from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import requests
import numpy as np
线程：爬取数据（IO）
def crawl_data(url):
resp = requests.get(url)
return resp.json()["data"]
进程：处理数据（CPU）
def process_data(data_list):
arr = np.array(data_list)
return arr.mean() 数值计算
if __name__ == "__main__":
1. 多线程爬取数据
urls =
with ThreadPoolExecutor(max_workers=5) as t_executor:
data = list(t_executor.map(crawl_data, urls))
2. 多进程处理数据
with ProcessPoolExecutor() as p_executor:
result = p_executor.submit(process_data, data).result()
print(f"数据均值：{result}")</pre></div>
頁:
[1]