import os

print('Process (%s) start...' % os.getpid())

# 只在 Unix/Linux/macOS 上有效
pid = os.fork()
if pid == 0:
    print('我是子进程 (%s)，父进程是 %s。' % (os.getpid(), os.getppid()))
else:
    print('我 (%s) 创建了子进程 (%s)。' % (os.getpid(), pid))

Windows 没有 fork()，所以这段代码在 Windows 上跑不了。这也引出了一个问题：怎么写跨平台的多进程程序？

2.2 multiprocessing 模块

multiprocessing 模块是 Python 提供的跨平台多进程方案，不用关心底层是 fork 还是别的实现。

2.2.1 创建 Process

Process 类代表一个进程对象，传入目标函数和参数，调用 start() 就能启动。

from multiprocessing import Process
import os


def run_proc(name):
    print('子进程 %s 在运行，PID: %s。' % (name, os.getpid()))


if __name__ == '__main__':
    print('父进程 PID: %s。' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('子进程即将启动...')
    p.start()
    p.join()
    print('子进程结束。')

运行结果：

父进程 PID: 928。
子进程即将启动...
子进程 test 在运行，PID: 929。
子进程结束。

start()：启动子进程。
join()：等待子进程结束，再往下执行。常用于进程间同步。

Windows 下运行 multiprocessing 时，一定要记得把创建进程的代码放到 if __name__ == '__main__': 块里，否则会报错。

2.2.2 进程池 Pool

如果需要批量创建很多子进程，用 Pool 更方便。它会自动管理一组进程，控制同时运行的数量。

from multiprocessing import Pool
import os, time, random


def long_time_task(name):
    print('运行任务 %s，PID: %s...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('任务 %s 运行了 %.2f 秒。' % (name, end - start))


if __name__ == '__main__':
    print('父进程 PID: %s。' % os.getpid())
    p = Pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print('等待所有子进程完成...')
    p.close()
    p.join()
    print('所有子进程已完成。')

运行结果：

父进程 PID: 669。
等待所有子进程完成...
运行任务 0，PID: 671...
运行任务 1，PID: 672...
运行任务 2，PID: 673...
运行任务 3，PID: 674...
任务 2 运行了 0.14 秒。
运行任务 4，PID: 673...
任务 1 运行了 0.27 秒。
任务 3 运行了 0.86 秒。
任务 0 运行了 1.41 秒。
任务 4 运行了 1.91 秒。
所有子进程已完成。

几个关键点：

Pool(4) 表示同时最多跑 4 个进程。
任务 0、1、2、3 立即执行，任务 4 要等前面某个任务完成才能上。
close() 之后就不能再添加新任务了。
join() 之前必须先调用 close()，否则报错。
默认 Pool 的大小等于 CPU 核数。

2.2.3 进程间通信

进程之间内存隔离，要交换数据需要专门的机制。multiprocessing 提供了 Queue 和 Pipes 等方式。

Queue 是最常用的：一个进程往里放数据，另一个进程从里面取。

from multiprocessing import Process, Queue
import os, time, random


def write(q):
    print('写进程 PID: %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print('写入 %s 到队列...' % value)
        q.put(value)
        time.sleep(random.random())


def read(q):
    print('读进程 PID: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('从队列读取 %s。' % value)


if __name__ == '__main__':
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))

    pw.start()
    pr.start()

    pw.join()
    pr.terminate()  # 读进程是死循环，只能强行终止

运行结果：

写进程 PID: 50563
写入 A 到队列...
读进程 PID: 50564
从队列读取 A。
写入 B 到队列...
从队列读取 B。
写入 C 到队列...
从队列读取 C。

在 Windows 下，multiprocessing 会先把对象通过 pickle 序列化再传给子进程。如果运行失败，先检查是不是对象不能被 pickle。

3. 多线程

3.1 创建线程

Python 提供了两个线程模块：_thread（低级）和 threading（高级）。日常用 threading 就够了。

import time, threading


def loop():
    print('线程 %s 正在运行...' % threading.current_thread().name)
    n = 0
    while n < 5:
        n += 1
        print('线程 %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('线程 %s 结束。' % threading.current_thread().name)


print('线程 %s 正在运行...' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('线程 %s 结束。' % threading.current_thread().name)

每个进程默认有一个主线程，名字是 MainThread。新线程的名字可以指定，不指定就自动叫 Thread-1、Thread-2……

threading.current_thread()：返回当前线程实例。
start()：启动线程。
join()：等待线程结束。

3.2 线程锁 Lock

多线程最大的麻烦是：所有线程共享同一个进程的变量。如果多个线程同时修改一个变量，结果很容易乱套。

看个例子：

import time, threading

balance = 0


def change_it(n):
    global balance
    balance = balance + n
    balance = balance - n


def run_thread(n):
    for i in range(1000000):
        change_it(n)


t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)

理论上，balance 先加后减，最终结果应该是 0。但由于线程调度不可控，balance = balance + n 这类语句实际上会拆成多步执行，中间可能被别的线程打断，结果不一定是 0。

解决办法就是加锁：

balance = 0
lock = threading.Lock()


def run_thread(n):
    for i in range(1000000):
        lock.acquire()
        try:
            change_it(n)
        finally:
            lock.release()

lock.acquire()：获取锁，只有一个线程能拿到。
lock.release()：释放锁，别人才能继续抢。
用 try...finally 保证锁一定会释放，否则可能造成死锁。

锁能保证一段关键代码同时只有一个线程执行，但也会降低并发效率。另外，多个线程各拿一把锁，互相等待对方释放，会造成死锁，程序卡死。

3.3 GIL 全局解释器锁

Python 的多线程还有一个绕不开的问题：GIL（Global Interpreter Lock）。

GIL 是 CPython 解释器的一个全局锁，任何线程执行 Python 字节码前，必须先拿到 GIL。解释器每执行一定数量的字节码，会自动释放 GIL，让其他线程有机会执行。

这意味着：Python 的多线程，不管有多少核，同一时刻只能有一个线程在执行 Python 字节码。所以多线程不能真正利用多核 CPU 来加速计算密集型任务。

但如果是 IO 密集型任务（比如网络请求、文件读写），线程大部分时间都在等 IO，GIL 的影响就没那么明显。

如果想利用多核，用多进程：每个进程有自己的 GIL，互不干扰。

4. ThreadLocal 线程局部变量

多线程里，有些数据希望每个线程各自有一份，互不影响。用局部变量虽然安全，但层层传递很麻烦；用全局变量又得加锁。

threading.local() 创建的对象，看起来像全局变量，但每个线程访问的都是自己的独立副本。

import threading

local_school = threading.local()


def process_student():
    # 获取当前线程关联的 student
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))


def process_thread(name):
    # 绑定当前线程的 student
    local_school.student = name
    process_student()


t1 = threading.Thread(target=process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target=process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

运行结果：

Hello, Alice (in Thread-A)
Hello, Bob (in Thread-B)

ThreadLocal 常用于绑定数据库连接、HTTP 请求、用户登录信息等，让同一个线程内的所有函数都能方便访问，又不用手动传递参数。

5. 进程 vs 线程

实现多任务，通常用 Master-Worker 模型：一个 Master 负责分配任务，多个 Worker 负责执行。用进程实现，Master 就是主进程，Worker 是子进程；用线程实现，Master 就是主线程，Worker 是子线程。

对比项	多进程	多线程
稳定性	高，子进程崩溃不影响其他进程	低，一个线程崩溃可能导致整个进程挂掉
创建开销	大	小
数据共享	难，需要 Queue/Pipes 等机制	容易，共享内存
多核利用	可以，每个进程有独立 GIL	受 GIL 限制，无法真正并行
适用平台	Unix/Linux 下效率高，Windows 下开销大	Windows 下效率相对更高

5.1 计算密集型 vs IO 密集型

任务还可以分成两类：

计算密集型：大量 CPU 运算，比如圆周率计算、视频解码。这类任务适合用 C/C++ 写，Python 本身效率不高。用多进程可以真正利用多核。
IO 密集型：大量网络、磁盘 IO，比如 Web 请求、文件读写。这类任务 CPU 消耗少，大部分时间都在等 IO。用多线程或者异步 IO 更合适，Python 也胜任。

5.2 异步 IO 的趋势

现代操作系统支持异步 IO，可以让一个线程同时处理多个 IO 任务。Python 里对应的就是协程，通过 asyncio 实现。后面会专门讲，这里先留个印象。

6. 分布式进程

multiprocessing 的 managers 子模块还能把进程分布到多台机器上。服务进程作为调度者，把任务通过网络分发到其他机器执行，Worker 进程处理完再把结果传回来。

简单思路就是把原来的 Queue 通过网络暴露出去：

task_master.py 注册任务队列和结果队列，监听端口，往任务队列里放任务。
task_worker.py 连接到服务进程，从任务队列取任务，计算完把结果放回结果队列。

# task_master.py 简版
from multiprocessing.managers import BaseManager
import queue, random

task_queue = queue.Queue()
result_queue = queue.Queue()

class QueueManager(BaseManager):
    pass

QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)

manager = QueueManager(address=('', 5000), authkey=b'abc')
manager.start()

task = manager.get_task_queue()
result = manager.get_result_queue()

for i in range(10):
    n = random.randint(0, 10000)
    task.put(n)

for i in range(10):
    print('结果: %s' % result.get(timeout=10))

manager.shutdown()

# task_worker.py 简版
from multiprocessing.managers import BaseManager
import queue, time

class QueueManager(BaseManager):
    pass

QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
manager.connect()

task = manager.get_task_queue()
result = manager.get_result_queue()

for i in range(10):
    try:
        n = task.get(timeout=1)
        r = '%d * %d = %d' % (n, n, n * n)
        time.sleep(1)
        result.put(r)
    except queue.Empty:
        print('任务队列为空。')

print('worker 结束。')

注意：

authkey 要保证主从一致，否则连不上。
队列里尽量传任务描述，而不是传大文件本身。比如处理日志，可以传文件路径，让 Worker 自己去读。