python 多进程和多线程

一、多进程

​ Unix/Linux系统提供了一个fork()系统调用，fork()调用一次，返回两次，因为操作系统自动把当前进程（父进程）复制了一份（子进程），然后，分别在父进程和子进程内返回。

​ 子进程永远返回0，而父进程返回子进程的ID。因此 A ，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。

​ Python的os模块封装了常见的系统调用，其中就包括fork，可以在Python程序中轻松创建子进程：

import os

print('进程 (%s) start...' % os.getpid())
# 要在 Unix/Linux/Mac操作系统中
pid = os.fork() #创建进程
if pid == 0:
    print('子进程： (%s) ，父进程： %s.' % (os.getpid(), os.getppid())) #getpid获取子进程ID，getppid获取父进程ID
else:
    print('(%s) 创建一个子进程 (%s).' % (os.getpid(), pid))

运行结果如下：

进程：(1876) start...  #ID每次都可能不一样，所以，不是1876也正常
(1876) 创建一个子进程 (1877).
子进程： (1877) ，父进程： 1876.

由于Windows没有fork调用，上面的代码在Windows上无法运行。

​ 有了fork调用，一个进程在接到新任务时就可以复制出一个子进程来处理新任务，常见的Apache服务器就是由父进程监听端口，每当有新的HTTP请求时，就fork出子进程来处理新的HTTP请求。

multiprocessing

​ Python中的multiprocessing模块是跨平台版本的多进程模块，在window上可以用multiprocessing模块编写多进程程序。

multiprocessing模块提供了一个Process类来代表一个进程对象，下面的例子演示了启动一个子进程并等待其结束：

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
    print('运行子进程： %s (%s)...' % (name, os.getpid()))

if __name__=='__main__':
    print('父进程 %s.' % os.getpid())
    p = Process(target=run_proc, args=('test',)) # run_proc: 函数名, args: 参数：输入参数后记得多一个,
    print('子进程将启动')
    p.start() #启动
    p.join()  # 让主进程等着，所有子进程执行完毕后，主进程才继续执行；注：卡住的是主进程，非子进程
    print('子进程结束')

执行结果如下：

父进程 5352.
子进程将启动
运行子进程： test (12004)...
子进程结束

join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。

Pool 进程池

如果要启动大量的子进程，可以用进程池的方式批量创建子进程：

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print('运行任务 %s (%s)...' % (name, os.getpid()))
    start = time.time() # 记录开始时间
    time.sleep(random.random() * 10) # 睡眠时间 1~10
    end = time.time() # 记录结束时间
    print('任务 %s 运行了 %0.2f 秒.' % (name, (end - start))) #打印运行时间，保留两位小数

if __name__=='__main__':
    print('父进程: %s.' % os.getpid())
    p = Pool(6) # 最大进程数
    for i in range(9):
        p.apply_async(long_time_task, args=(i,))
    print('正在等待所有子流程完成')
    p.close() #关闭进程插入：只是不能再添加进程，不是把正在运行的进程关闭
    p.join() 
    print('所有子流程都已完成')

执行结果如下：

父进程: 14860.
正在等待所有子流程完成
运行任务 0 (17804)...
运行任务 1 (16432)...
运行任务 2 (16136)...
运行任务 3 (16380)...
运行任务 4 (17916)...
运行任务 5 (2828)...
任务 1 运行了 0.57 秒.
运行任务 6 (16432)...
任务 4 运行了 0.74 秒.
运行任务 7 (17916)...
任务 2 运行了 2.71 秒.
运行任务 8 (16136)...
任务 5 运行了 3.75 秒.
任务 8 运行了 2.70 秒.
任务 7 运行了 5.50 秒.
任务 6 运行了 5.78 秒.
任务 0 运行了 6.73 秒.
任务 3 运行了 9.06 秒.
所有子流程都已完成

​ 请注意输出的结果，task 0，1，2，3，4，5，6是立刻执行的，而task 7, 8要等待前面某个task完成后才执行，这是因为Pool的默认大小在我的电脑上是6，因此，最多同时执行6个进程。这是Pool有意设计的限制，并不是操作系统的限制。如果改成：

p = Pool(9)

就可以同时跑9个进程。

由于Pool的默认大小是CPU的核数，如果你不幸拥有12核CPU，你要提交至少13个子进程才能看到上面的等待效果。

子进程

子进程在很多时候并不是自身，而是一个外部进程，需要我们进行控制。subprocess模块可以让我们非常方便地启动一个子进程，然后控制其输入和输出。

下面的例子演示了如何在Python代码中运行命令nslookup www.baidu.com，这和命令行直接运行的效果是一样的：

import subprocess

print('$ nslookup www.baidu.com')
r = subprocess.call(['nslookup', 'www.baidu.com'])
print('Exit code:', r) #返回结果 0成功，1和2都是错误执行，2通常是没有读取到文件，1的反馈目前未知

运行结果：

$ nslookup www.a.shifen.com
Server:		192.168.19.4
Address:	192.168.19.4#53

Non-authoritative answer:
www.a.shifen.com	
Name:	www.a.shifen.com
Address: 14.215.177.39

Exit code: 0

如果子进程还需要输入，则可以通过communicate()方法输入：

import subprocess

print('$ nslookup')
p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b'set q=mx\npython.org\nexit\n')
print(output.decode('gbk'))
print('Exit code:', p.returncode)

上面的代码相当于在命令行执行命令nslookup，然后手动输入：

set q=mx
python.org
exit

运行结果如下：

$ nslookup
默认服务器:  cache-a.guangzhou.gd.cn
Address:  202.96.128.86

> > 服务器:  cache-a.guangzhou.gd.cn
Address:  202.96.128.86

python.org	MX preference = 50, mail exchanger = mail.python.org

python.org	nameserver = ns-1134.awsdns-13.org
python.org	nameserver = ns-981.awsdns-58.net
python.org	nameserver = ns-484.awsdns-60.com
python.org	nameserver = ns-2046.awsdns-63.co.uk
mail.python.org	internet address = 188.166.95.178
ns-2046.awsdns-63.co.uk	internet address = 205.251.199.254
ns-484.awsdns-60.com	internet address = 205.251.193.228
ns-981.awsdns-58.net	internet address = 205.251.195.213
mail.python.org	AAAA IPv6 address = 2a03:b0c0:2:d0::71:1
ns-1134.awsdns-13.org	AAAA IPv6 address = 2600:9000:5304:6e00::1
ns-2046.awsdns-63.co.uk	AAAA IPv6 address = 2600:9000:5307:fe00::1
ns-484.awsdns-60.com	AAAA IPv6 address = 2600:9000:5301:e400::1
ns-981.awsdns-58.net	AAAA IPv6 address = 2600:9000:5303:d500::1
> 
Exit code: 0

进程间通信

Process之间肯定是需要通信的，操作系统提供了很多机制来实现进程间的通信。Python的multiprocessing模块包装了底层的机制，提供了Queue、Pipes等多种方式来交换数据。

我们以Queue为例，在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：

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

# 写数据进程执行的代码:
def write(q):
    print('Process to write: %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print('Put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())

# 读数据进程执行的代码:
def read(q):
    print('Process to read: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('Get %s from queue.' % value)

if __name__=='__main__':
    # 父进程创建Queue，并传给各个子进程：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 启动子进程pw，写入:
    pw.start()
    # 启动子进程pr，读取:
    pr.start()
    # 等待pw结束:
    pw.join()
    # pr进程里是死循环，无法等待其结束，只能强行终止:
    pr.terminate()

运行结果如下：

Process to write: 18356
Put A to queue...
Process to read: 18104
Get A from queue.
Put B to queue...
Get B from queue.
Put C to queue...
Get C from queue.

​ 在Unix/Linux下，multiprocessing模块封装了fork()调用，所以我们不需要关注fork()的细节。由于Windows没有fork调用，因此，multiprocessing需要“模拟”出fork的效果，父进程所有Python对象都必须通过pickle序列化再传到子进程去，所以，如果multiprocessing在Windows下调用失败了，要先考虑是不是pickle失败了。

参考源码

do_folk.py multi_processing.py pooled_processing.py do_subprocess.py do_queue.py

二、多线程

​ 多任务可以由多进程完成，也可以由一个进程内的多线程完成。我们前面提到了进程是由若干线程组成的，一个进程至少有一个线程。

​ 由于线程是操作系统直接支持的执行单元，因此，高级语言通常都内置多线程的支持，Python也不例外，并且，Python的线程是真正的Posix Thread，而不是模拟出来的线程。

​ Python的标准库提供了两个模块：_thread和threading，_thread是低级模块，threading是高级模块，对_thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。

​ 启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()开始执行：

import time, threading

# 新线程执行的代码:
def loop():
    print('thread %s is running...' % threading.current_thread().name)
    for n in range(1, 6):
        print('thread %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('thread %s ended.' % threading.current_thread().name)

print('thread %s is running...' % threading.current_thread().name) # 主线程
t = threading.Thread(target=loop, name='TestThread') # target: 函数名， name: 线程名，除打印外无意义，可以不填
t.start()  # 开始线程
t.join() # 和进程一样意思
print('thread %s ended.' % threading.current_thread().name)

执行结果如下：

thread MainThread is running...
thread TestThread is running...
thread TestThread >>> 1
thread TestThread >>> 2
thread TestThread >>> 3
thread TestThread >>> 4
thread TestThread >>> 5
thread TestThread ended.
thread MainThread ended.

​ 由于任何进程默认就会启动一个线程，我们把该线程称为主线程，主线程又可以启动新的线程，Python的threading模块有个current_thread()函数，它永远返回当前线程的实例。主线程实例的名字叫MainThread，子线程的名字在创建时指定，名字仅仅在打印时用来显示，完全没有其他意义，如果不起名字Python就自动给线程命名为Thread-1，Thread-2……

Lock锁

​ 多线程和多进程最大的不同在于，多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响，而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。来看看多个线程同时操作一个变量怎么把内容给改乱了：

​ 我们定义了一个共享变量test，初始值为0，并且启动两个线程，先存后取，理论上结果应该为0，但是，由于线程的调度是由操作系统决定的，当t1、t2交替执行时，只要循环次数足够多，test的结果就不一定是0了。

test = test + n

分两步：

1. 计算test + n，存入临时变量中；
2. 将临时变量的值赋给test。

也就是可以看成：

x = test + n
test = x

由于x是局部变量，两个线程各自都有自己的x，当代码正常执行时：

初始值 test = 0

t1: x1 = test + 1 # x1 = 0 + 1 = 1
t1: test = x1     # test = 1
t1: x1 = test - 1 # x1 = 1 - 1 = 0
t1: test = x1     # test = 0

t2: x2 = test + 8 # x2 = 0 + 8 = 8
t2: test = x2     # test = 8
t2: x2 = test - 8 # x2 = 8 - 8 = 0
t2: test = x2     # test = 0
    
结果 test = 0

但是t1和t2是交替运行的，如果操作系统以下面的顺序执行t1、t2：

初始值 test = 0

t1: x1 = test + 1  # x1 = 0 + 1 = 1

t2: x2 = test + 8  # x2 = 0 + 8 = 8
t2: test = x2      # test = 8

t1: test = x1      # test = 1
t1: x1 = test - 1  # x1 = 1 - 1 = 0
t1: test = x1      # test = 0

t2: x2 = test - 8  # x2 = 0 - 8 = -8
t2: test = x2      # test = -8

结果 test = -8

​ 究其原因，是因为修改test需要多条语句，而执行这几条语句时，线程可能中断，从而导致多个线程把同一个对象的内容改乱了。

两个线程同时一存一取，就可能导致数值不对，所以，我们必须确保一个线程在修改test的时候，别的线程一定不能改。

​ 如果我们要确保test计算正确，就要给a()上一把锁，当某个线程开始执行a()时，我们说，该线程因为获得了锁，因此其他线程不能同时执行a()，只能等待，直到锁被释放后，获得该锁以后才能改。由于锁只有一个，无论多少线程，同一时刻最多只有一个线程持有该锁，所以，不会造成修改的冲突。创建一个锁就是通过threading.Lock()来实现：

import threading

def a():
    global A,lock
    lock.acquire()
    try:
        for i in range(10):
            A+=1
            print("a",A)
    finally:
        lock.release()
        
def b():
    global A,lock
    lock.acquire()
    try:
        for i in range(10):
            A+=10
            print("b",A)
    finally:
    	lock.release()

if __name__ == '__main__':
    lock = threading.Lock()
    A=0
    t1=threading.Thread(target=a,)
    t2=threading.Thread(target=b,)
    t1.start()
    t2.start()

​ 当多个线程同时执行lock.acquire()时，只有一个线程能成功地获取锁，然后继续执行代码，其他线程就继续等待直到获得锁为止。获得锁的线程用完后一定要释放锁，否则那些苦苦等待锁的线程将永远等待下去，成为死线程。所以我们用try...finally来确保锁一定会被释放。

多核CPU

​ 如果你拥有一个多核CPU，你肯定在想，多核应该可以同时执行多个线程。如果写一个死循环的话，会出现什么情况呢？

​ 我们可以监控到一个死循环线程会100%占用一个CPU。如果有两个死循环线程，在多核CPU中，可以监控到会占用200%的CPU，也就是占用两个CPU核心。要想把N核CPU的核心全部跑满，就必须启动N个死循环线程。

试试用Python写个死循环：

import threading, multiprocessing

def loop():
    x = 0
    while True:
        x = x ^ 1

for i in range(multiprocessing.cpu_count()):
    t = threading.Thread(target=loop)
    t.start()

​ 启动与CPU核心数量相同的N个线程，在4核CPU上可以监控到CPU占用率仅有102%，也就是仅使用了一核。但是用C、C++或Java来改写相同的死循环，直接可以把全部核心跑满，4核就跑到400%，8核就跑到800%。因为Python的线程虽然是真正的线程，但解释器执行代码时，有一个GIL锁：Global Interpreter Lock，任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，再让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都上了锁，所以，多线程在Python中只能交替执行，即使50个线程跑在50核CPU上，也只能用到1个核。所以，在Python中，可以使用多线程，但不要指望能有效利用多核。如果一定要通过多线程利用多核，那只能通过C扩展来实现，不过这样就失去了Python简单易用的特点。不过，Python虽然不能利用多线程实现多核任务，但可以通过多进程实现多核任务。多个Python进程有各自独立的GIL锁，互不影响。

参考源码

multi_threading.py do_lock.py