进程
前言
- 这一篇讲
multiprocessing：怎么开进程、Windows 上「启动方式」带来的坑、进程锁、进程池、进程间通信（Pipe/Queue）、共享状态（Manager/Value/Array/shared_memory），以及自定义进程类。 - 所有示例在本机 Python 3.14.10（Windows）上实跑过，输出原样贴在代码块下面。为了让输出可比对，示例里不直接打印 pid，而是打印「pid 是否不同」这类稳定结论。
进程简介
- 进程：操作系统进行资源分配的基本单位，一个进程有自己独立的地址空间（地址空间 = 这套进程独占的一整片内存编号，别的进程看不见也改不了里面的东西）。在 Python 里，一个进程对应一个解释器实例、一份独立的内存。
- 线程：CPU 调度的基本单位，同进程内的线程共享地址空间（细节见《线程》）。
- 单核怎么「多任务」：把 CPU 时间切成小片轮流分配给任务，切换足够快，人眼看不出而已。
- 多核：任务数不超过核心数时可以真正并行；实际系统里任务远多于核心，仍然是轮转，只是「干活的人」多了几个。
- 为什么要用多进程而不是多线程：CPython 有 GIL（全局解释器锁，见《线程》），纯 Python 字节码（源码先编译成的中间指令，解释器再逐条执行）执行同一时刻只有一个线程在跑；要吃满多核，计算密集型任务得用多进程（或换解释器 / C 扩展释放 GIL，3.13 起还有实验性的 free-threaded 构建）。
- 一个进程能看到的 CPU 时间片由操作系统决定，Python 只负责创建和管理进程对象。
Process 的基本用法
Process(target=函数, args=(...), kwargs={...}, name=...) 只是创建对象，要 start() 才真正起进程；join() 等它结束。
Process() 的签名是 Process(group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)，参数逐个说：
target：子进程要执行的函数对象，写函数名不要加括号（child而不是child()）。不传也行，子进程空跑一趟、exitcode仍是0。target的函数必须定义在模块顶层：Windows 的spawn要把目标函数 pickle 后带过去，lambda 会PicklingError: Can't pickle local object，写在if __name__里的嵌套函数会在子进程报AttributeError: module '__mp_main__' has no attribute '函数名'。
args：传给target的位置参数，必须是元组，默认()。只有一个参数也得带逗号：args=(main_pid,)；写成args=(1)会在start()当场TypeError: 'int' object is not iterable。kwargs：传给target的关键字参数，字典，默认None（当作{}）。args/kwargs里的每个值都要能被 pickle，连接对象、锁句柄这类传不过去，得改用Queue/Manager（见后面「进程间通信」）。name：进程名，纯用于辨认——p.name、multiprocessing.active_children()、日志里靠它区分是谁。daemon：*之后的关键字限定参数，只能写成daemon=True，等价于创建后再设p.daemon = True。含义是「父进程退出时我被一起带走」，详见下面守护进程那节；只能在start()之前设，启动之后改会AssertionError: process has already started。group：历史遗留参数，必须留None（Python 3.9 起废弃），传别的值直接AssertionError: group argument must be None for now。
默认名 Process-N 里的 N 按创建顺序递增，手动命名的也占号——下面这段里先建了两个对象，所以后面两个默认名是 Process-3、Process-4，它和 pid 没有任何关系：
1 | import inspect |
输出：
1 | 签名: (self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None) |
1 | import os |
输出：
1 | 主进程 pid 是正整数: True |
start() 和 join() 各自到底做了什么（这两个名字最容易「记住叫法、没记住行为」）：
start()：向操作系统申请创建子进程，并让新进程从target指定的那个函数开始跑。它是「开始」不是「跑完」——调用完立刻返回，父进程继续执行下一行，于是父子并发，谁先打印不固定。同一个对象只能start()一次，重复调用报AssertionError: cannot start a process twice。join()：让调用它的一方（这里是父进程）停在原地等，等那个子进程结束才往下走。名字容易误解成「把两个进程合并」，实际意思是「我等它」。返回值永远是None；join(0.2)是「最多等 0.2 秒」，到点就返回（可能还没等完，要配is_alive()判断）；没start()就join()报AssertionError: can only join a started process；重复join()不报错。join()等的是它那一个，不是「最先跑完的那个」：下面 B 先结束，p1.join()仍然要等 A 跑完才返回。- 等待期间父进程是挂起的（不是空转轮询），不占 CPU。
1 | import time |
输出：
1 | 两个 start 都调用完了，主进程立刻走到下一行 |
几个要点：
start()之后父进程和子进程并发执行，谁先打印不固定；join()才把顺序钉住。join()的返回值永远是None，它只负责等；带超时join(0.2)到点就返回，此时子进程可能还没结束，所以要配is_alive()判断有没有等完。is_alive()：start()之前False，运行中True，join()之后False。pid：操作系统进程号，start()之前是None。exitcode：None表示还在跑，0正常结束，os._exit(3)就是3。- 子进程里抛未捕获异常时
exitcode是1，traceback 只打在子进程自己的 stderr，父进程看不到——只能靠这个1发现出过事。 terminate()不等了直接终止子进程，之后仍要join()收尾；exitcode记成-15，Windows 上没有真信号，multiprocessing 用 SIGTERM 的编号表示「被终止」。- 传参一律走
args/kwargs，参数会被pickle后带进子进程。
1 | import os |
输出：
1 | 不传 target exitcode = 0 |
启动方式与「全局变量不共享」
-
三种启动方式
spawn全新起一个解释器并重新导入主模块。fork直接复制父进程内存（Unix）。forkserver由一个干净的服务进程派生（Unix）。Windows 和 macOS 上只有spawn。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import multiprocessing
import os
VAR = 100
def show():
# spawn 起来的子进程会重新导入主模块，VAR 在子进程里是「刚初始化」的值
print('子进程看到的 VAR =', VAR)
if __name__ == '__main__':
print('当前启动方式:', multiprocessing.get_start_method())
print('可用的启动方式:', sorted(multiprocessing.get_all_start_methods()))
VAR = 999 # 主进程里改了，子进程看不到
p = multiprocessing.Process(target=show)
p.start()
p.join()
print('主进程里的 VAR =', VAR)输出：
1
2
3
4当前启动方式: spawn
可用的启动方式: ['spawn']
子进程看到的 VAR = 100
主进程里的 VAR = 999子进程打印的
VAR是 100 而不是 999：因为spawn会重新导入主模块，模块级代码又执行了一遍VAR = 100，而if __name__ == '__main__':那块在子进程里不会执行。这也解释了「为什么必须把启动代码写进
if __name__ == '__main__':」——不写的话，子进程重新导入时又会执行一遍start()，无限套娃（RuntimeError或直接把机器打满）。 -
传值要用参数
1
2
3
4
5
6
7
8
9
10
11import os
from multiprocessing import Process
def child(main_pid):
# 想跨进程传值就用参数（会被 pickle 后带过去）
print('通过参数拿到主进程 pid:', os.getpid() != main_pid)
if __name__ == '__main__':
p = Process(target=child, args=(os.getpid(),))
p.start()
p.join()输出：
1
通过参数拿到主进程 pid: True
-
全局变量改不动对方
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17import multiprocessing
num = 250
lt = ['hello']
def child():
global num, lt
num += 10
lt.append('world')
print('子进程里:', num, lt)
if __name__ == '__main__':
print('主进程改之前:', num, lt)
p = multiprocessing.Process(target=child)
p.start()
p.join()
print('主进程改之后:', num, lt, '<- 子进程的修改不会回来')输出：
1
2
3主进程改之前: 250 ['hello']
子进程里: 260 ['hello', 'world']
主进程改之后: 250 ['hello'] <- 子进程的修改不会回来num是 int、lt是 list，两种都一样：子进程拿到的是副本，改完不会回传。想共享得用下一节的机制。
守护进程与子进程异常
daemon = True（或Process(..., daemon=True)）表示「父进程一退出我就被带走」；不设的话，Python 进程退出时会等待非守护子进程结束。- 「守护」这个词容易让人以为它在替谁值班——其实谁也没守护。它来自 Unix 里长期在后台跑的服务程序（daemon，如
sshd、mysqld），这类程序的共同点是「依附于上级，上级走了它就该消失」。Python 借这个词表达的就是这层依附关系：守护子进程不要求收尾，父进程退出时它被直接终止。 - 两条实测出来的规矩：
- 守护进程不许再生子进程：在里面
Process(...).start()会抛AssertionError: daemonic processes are not allowed to have children（想套娃就让中间那层非守护）。 - 被带走是硬终止，
finally不保证执行：实测守护子进程只打印了「开始」，finally那行没出现；同一份代码把daemon=True改成False，finally就正常执行了。所以「一定要写完」的活别交给守护进程，要么非守护 +join，要么用Pool的close()+join()收尾。
- 守护进程不许再生子进程：在里面
- 一个很自然的误解：看到「守护」两个字，容易理解成「它守护着孩子——主进程退出了，子进程被保护着继续跑完」。实际规则正相反，
daemon=True是「孩子跟着大人走」。 - 名字为什么这么绕：Unix 的 daemon 指「脱离终端、长期在后台活着的服务进程」；Python 的
daemon=True指「依附上级，上级走了我就不必收尾」。同一个词，生命周期规则正好相反——这也是最容易记反的地方。
1 | import time |
输出：
1 | 守护进程开始 |
主进程 1.5 秒后就走，long_job 里的 time.sleep(5) 等不到，第二句不会被打印——这就是 daemon 的语义。两个细节：示例里的 print(..., flush=True) 不是可有可无，子进程被强杀时缓冲区里的输出会丢；反过来，daemon 进程不能再创建子进程（daemonic processes are not allowed to have children）。
子进程里抛异常只会结束它自己：
1 | from concurrent.futures import ProcessPoolExecutor |
输出：
1 | 主进程收到子进程的异常: 子进程里抛异常 |
（子进程的 traceback 会先打到 stderr，被 join() 收回来的是退出码。）想拿到异常内容，用 concurrent.futures.ProcessPoolExecutor 更省事——它把异常原样抛回父进程。
进程锁
多个进程写同一个资源（共享计数、同一个文件）会互相踩，用 Lock 把临界区串行化。
1 | import multiprocessing |
输出：
1 | 任务 4 完成，当前计数 1 |
- 计数一路 1→5 不跳号，说明每次
+= 1都在独占状态下完成；不加锁时「读-改-写」可能被别的进程插队，最终结果就会小于 5。 with lock:会自动acquire/release，比手写lock.acquire()安全（中途抛异常也能释放）。- 还有
RLock（可重入）、Semaphore（限流）、Condition（等待/通知），用法和threading里的同名对象一致，只是跨进程。 multiprocessing.Lock的接口和threading.Lock一样（acquire(blocking=True, timeout=-1)/release()），但有一条必须记住的差别：它得作为参数传给子进程（Process(target=work, args=(lock, counter))），不能像线程那样靠全局变量拿——子进程是独立内存，全局变量在那边是重新导入出来的另一份。Pool场景里则要用Pool(initializer=..., initargs=(lock,))传进去。- 写文件更常见的做法是「各写各的临时文件，最后合并」，锁只适合临界区很短的场景。
进程池
任务多、要控制并发数时，用 Pool 而不是手搓一堆 Process。
1 | import multiprocessing |
输出：
1 | CPU 核心数是正整数: True |
map会按入参顺序返回结果，即使完成顺序不同。imap是迭代器（不一次性把全部结果装进内存，而是每次要下一个时才现算），边算边给，适合结果很大或想早点开始处理。apply_async返回AsyncResult，用.get(timeout=...)取值；一定要给 timeout，否则子进程卡死时父进程会永远等下去。- 池大小一般不超过
cpu_count()；with Pool(n) as pool退出时会自动terminate，正常收尾仍推荐close()+join()。
1 | import multiprocessing |
输出：
1 | map 保序: ['ALPHA', 'BETA', 'GAMMA', 'DELTA'] |
chunksize 决定「一次给每个进程派多少个任务」：任务多且每个很轻时调大能省 IPC 开销；任务少且重就用默认。
回调写法：
1 | import multiprocessing |
输出：
1 | 回调收集到（排序后）: [100, 101, 102] |
回调在父进程里执行，参数是任务函数的返回值；三个回调的先后顺序不固定（取决于谁先算完），但 close() + join() 能保证它们都在「主进程结束」之前跑完。
进程间通信：Pipe 与 Queue
-
Pipe：两端直连
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import multiprocessing
def child(conn):
data = conn.recv()
print('子进程收到:', data)
conn.send([x * 2 for x in data])
if __name__ == '__main__':
parent, child_end = multiprocessing.Pipe()
p = multiprocessing.Process(target=child, args=(child_end,))
p.start()
parent.send([1, 2, 3])
print('主进程收到回执:', parent.recv())
p.join()
read_end, write_end = multiprocessing.Pipe(duplex=False)
write_end.send('半双工')
print('duplex=False 读端拿到:', read_end.recv())
print('写端还有 recv 方法吗:', hasattr(write_end, 'recv'))
try:
write_end.recv()
except OSError as e:
print('真的去读会:', type(e).__name__, e)输出：
1
2
3
4
5子进程收到: [1, 2, 3]
主进程收到回执: [2, 4, 6]
duplex=False 读端拿到: 半双工
写端还有 recv 方法吗: True
真的去读会: OSError connection is write-only- 默认全双工，两端都能收发。
duplex=False时conn1只能读、conn2只能写（注意recv方法仍然存在，调用才会报错，所以hasattr是 True）。适合一对一。recv()在对端没发之前会阻塞（就是卡在这一行不返回，后面的代码一行都走不下去），两端都等就死锁。
-
Queue：多对多的任务分发
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22import multiprocessing
def consumer(q, out, tag):
got = [q.get() for _ in range(3)] # 队列空时这里会阻塞
out.extend(got)
if __name__ == '__main__':
q = multiprocessing.Queue(3) # maxsize=3：放满之后 put 会等
print('刚创建时 empty:', q.empty())
shared = multiprocessing.Manager().list()
c1 = multiprocessing.Process(target=consumer, args=(q, shared, 'A'))
c2 = multiprocessing.Process(target=consumer, args=(q, shared, 'B'))
c1.start()
c2.start() # 先让消费者就位，再投喂
for i in range(6):
q.put(i)
c1.join()
c2.join()
print('两个消费者合起来取到:', sorted(shared))
print('取空之后 empty:', q.empty())
q.cancel_join_thread()
q.close()输出：
1
2
3刚创建时 empty: True
两个消费者合起来取到: [0, 1, 2, 3, 4, 5]
取空之后 empty: True每个元素只会被一个消费者取走，所以两个消费者合起来正好拿到 0~5。要点：
put/get走的是后台线程 + 管道，qsize()、empty()、full()只是近似值，别拿它们做严格判断。- 队列空时
get()会阻塞，所以要么放「毒丸」（None）通知退出，要么用get(timeout=...)。 - 进程结束前记得
cancel_join_thread()/close()，否则join()可能挂住。 - 需要「生产者-消费者 + 有界缓冲」时，
Queue(maxsize=n)就是背压：满了put就等。
共享状态：Manager 与 shared_memory
-
Manager：什么都能共享，但慢
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import multiprocessing
def child(v, s, a, lst, dct):
v.value += 10
s.value = b'world'
a[0] = 5
lst.append('大哥，您收好！')
dct['name'] = 'xiaoming'
if __name__ == '__main__':
server = multiprocessing.Manager()
v = server.Value('i', 250)
s = server.Value('s', b'hello')
a = server.Array('i', range(5))
lst = server.list()
dct = server.dict()
p = multiprocessing.Process(target=child, args=(v, s, a, lst, dct))
p.start()
p.join()
print('主进程看到 v =', v.value)
print('主进程看到 s =', s.value)
print('主进程看到 a[0] =', a[0])
print('主进程看到 list =', list(lst))
print('主进程看到 dict =', dict(dct))
server.shutdown()输出：
1
2
3
4
5主进程看到 v = 260
主进程看到 s = b'world'
主进程看到 a[0] = 5
主进程看到 list = ['大哥，您收好！']
主进程看到 dict = {'name': 'xiaoming'}Manager()起一个服务进程，Value/Array/list/dict都是代理对象，每次读写都在走 IPC——所以它方便但慢，而且lst.append()这种修改对代理是有效的（和「全局变量不共享」正好相反）。
Value('i', 250)的类型码来自ctypes：i是 int、f是 float、s是c_char_p（要传 bytes）。 -
不加 Manager 的共享变量
multiprocessing.Value/Array直接放在共享内存里，比 Manager 快得多；需要原子性时给它配锁：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15import multiprocessing
def worker(v, n):
for _ in range(n):
with v.get_lock(): # Value 自带一把锁
v.value += 1
if __name__ == '__main__':
v = multiprocessing.Value('i', 0)
ps = [multiprocessing.Process(target=worker, args=(v, 1000)) for _ in range(4)]
for p in ps:
p.start()
for p in ps:
p.join()
print('4 个进程各加 1000 次，结果 =', v.value)输出：
1
4 个进程各加 1000 次，结果 = 4000
-
shared_memory：直接共享一大块字节
3.8 加入，适合数组类大数据，避免「每次读写都走 IPC」。
1
2
3
4
5
6
7
8
9
10
11import multiprocessing.shared_memory as shm
block = shm.SharedMemory(create=True, size=16)
print('块名是随机字符串:', len(block.name) > 8)
view = memoryview(block.buf)[:8]
view[0:4] = b'abcd'
print('写入后读回:', bytes(block.buf[:4]))
view.release()
block.close()
block.unlink()
print('unlink 之后资源已释放')输出：
1
2
3块名是随机字符串: True
写入后读回: b'abcd'
unlink 之后资源已释放它只是一块裸内存，需要自己配
numpy或ctypes解释内容，也要记得close()+unlink()（不 unlink 会残留到系统重启）。NumPy 数组跨进程共享就是shared_memory+np.frombuffer的组合。
自定义进程类
继承 Process 并实现 run()，start() 后自动执行；适合把「一个 worker 的状态和行为」封装起来。
1 | import time |
输出：
1 | 子进程运行中... |
注意 run() 是在子进程里执行的，self.delay、self.times 这些属性是随对象一起被 pickle 过去的；在 run() 里改的属性不会回到父进程。
整体示例：进程池 + 锁 + 共享计数
把锁、Value、Manager().list()、进程池串成一个能验证的小任务分发器：三个耗时不同的任务，池里最多 2 个进程并发，用锁保证计数和记录不串。
1 | import multiprocessing |
输出：
1 | 提交任务数: 3 |
读法：
Pool(2)只有两个工人，三个任务分两批跑，所以总耗时接近「最慢的那批」而不是三个之和。- 完成顺序由睡眠时长决定：
压缩日志(0.05) →发送通知(0.10) →下载图片(0.15)。 - 计数用锁保护，
out用 Manager 列表收集，最后按序号排序还原完成顺序——这是「并发执行、有序汇报」的常见套路。