程序卡住了？多进程多线程让你的 Python“多任务并行“ 跑起来

程序运行慢很多时候是它 只肯一次干一件事。这篇老哥带你搞懂并发 / 并行、进程 / 线程让程序真正 多任务跑起来。一、并发 vs 并行并发Concurrency单个 CPU 轮流处理多个任务每个任务交替执行一段时间并行Parallelism多个 CPU 同时执行多个任务。二、多进程资源分配的基本单位进程是操作系统资源分配的基本单位 —— 一个正在运行的程序就是一个进程。每个进程有独立的内存空间一个进程崩溃不影响其他进程。用multiprocessing.Process创建进程。注意 Windows 上要加if __name__ __main__import time import multiprocessing def write_file(): with open(test.txt, a) as f: while True: f.write(hello world\n) f.flush() time.sleep(0.5) def read_file(): with open(test.txt, r) as f: while True: time.sleep(0.1) print(f.read(1)) if __name__ __main__: p1 multiprocessing.Process(targetwrite_file) p2 multiprocessing.Process(targetread_file) p1.start() # 启动 p2.start()Process 常用属性 / 方法name名称、pid进程号、start()启动、join([timeout])阻塞主进程等子进程、terminate()强制终止、is_alive()。自定义 Process 子类重写run()import os import multiprocessing class Worker(multiprocessing.Process): def run(self): print(进程id, os.getpid(), 父进程id, os.getppid()) if __name__ __main__: for i in range(5): p Worker() p.start()三、进程池批量管理子进程子进程太多就要用进程池 Pool。apply()同步阻塞、apply_async()异步不阻塞import os import time import multiprocessing def func(): for i in range(10): print(os.getpid(), i) time.sleep(0.5) if __name__ __main__: pool multiprocessing.Pool(process_num5) # 指定进程池大小 for p in range(5): pool.apply_async(func) # 非阻塞式提交 pool.close() # 关闭提交 pool.join() # 等待所有任务完成 print(end)四、进程间通信Queue进程间不共享全局变量各自独立内存要传数据得用Queue队列先进先出import time import random import multiprocessing def func1(queue): # 生产 while True: queue.put(random.randint(1, 50)) time.sleep(random.random()) def func2(queue): # 消费 while True: print( * queue.get()) if __name__ __main__: queue multiprocessing.Queue() p1 multiprocessing.Process(targetfunc1, args(queue,)) p2 multiprocessing.Process(targetfunc2, args(queue,)) p1.start(); p2.start() p1.join(); p2.join()进程池之间用Manager().Queue()通信queue multiprocessing.Manager().Queue() pool multiprocessing.Pool(2) pool.apply_async(func1, (queue,)) pool.apply_async(func2, (queue,))五、多线程任务调度的基本单位线程是任务调度和执行的基本单位。一个进程至少一个线程。和进程不同多线程之间可共享数据—— 但线程出错且没捕获会导致整个进程崩溃。用threading.Thread创建线程import time import threading def func(): while True: print(threading.current_thread().name, 0000) time.sleep(0.5) if __name__ __main__: t1 threading.Thread(targetfunc, name线程1) t2 threading.Thread(targetfunc, name线程2) t1.start() t2.start()线程池 ThreadPoolExecutor来自concurrent.futuresimport threading import time import concurrent.futures def func(): for i in range(6): print(threading.current_thread().name, f数据{i}) time.sleep(0.2) if __name__ __main__: with concurrent.futures.ThreadPoolExecutor(max_workers3) as executor: future1 executor.submit(func) # 提交任务返回 Future future2 executor.submit(func) future3 executor.submit(func) future1.result() # 取结果六、线程安全与互斥锁线程共享数据会有线程安全问题。看这个 3 个线程各把 g_num 加 1 十次 的例子期望 30import time import threading def func(): global g_num for _ in range(10): tmp g_num 1 time.sleep(0.01) # 故意加延迟 g_num tmp print(f{threading.current_thread().name}: {g_num}\n, end) if __name__ __main__: g_num 0 threads [threading.Thread(targetfunc, namef线程{i}) for i in range(3)] [t.start() for t in threads] [t.join() for t in threads] print(g_num) # 结果可能是 10而不是 30原因线程 A 在tmp g_num 1后休眠还没来得及把g_num tmpCPU 就切给线程 BB 读到的还是旧值。这就是线程不安全。互斥锁解决同一时刻只有一个线程能改共享数据。def func(): global g_num for _ in range(10): lock.acquire() # 获取锁 tmp g_num 1 time.sleep(0.01) g_num tmp lock.release() # 释放锁 print(f{threading.current_thread().name}: {g_num}\n, end) if __name__ __main__: g_num 0 lock threading.Lock() # 创建锁 threads [threading.Thread(targetfunc, namef线程{i}) for i in range(3)] [t.start() for t in threads] [t.join() for t in threads] print(g_num) # 30 正确了七、GILPython 的 全局解释器锁GILGlobal Interpreter Lock是 CPython 实现引入的一个锁同一时间只允许一个线程持有 Python 解释器的控制权所以多线程无法真正并行利用多核 CPU。CPU 密集型任务大量计算GIL 是性能瓶颈多线程没优势该用多进程I/O 密集型任务文件读写、网络请求线程在 I/O 等待时会释放 GIL其他线程可以执行多线程依然有效。一句话计算密集 → 多进程I/O 密集 → 多线程。老哥收尾进程线程的选择口诀进程独立内存、线程共享数据计算密集用多进程绕开 GILI/O 密集用多线程共享数据记得上互斥锁大量任务用池Pool / ThreadPoolExecutor。到这里Python 的 多任务 也打通了。下一篇讲网络编程—— 让程序能联网通信。有问题随时回来找老哥。