python 多线程并行编程详细讲解

三、用python写并行程序

在介绍如何使用python写并行程序之前，我们需要先补充几个概念，分别是进程、线程与全局解释器锁（Global Interpreter Lock, GIL）。

3.1 进程与线程

进程（process）：

• 在面向线程设计的系统（如当代多数操作系统、Linux 2.6及更新的版本）中，进程本身不是基本运行单位，而是线程的容器。
• 进程拥有自己独立的内存空间，所属线程可以访问进程的空间。
• 程序本身只是指令、数据及其组织形式的描述，进程才是程序的真正运行实例。 例如，Visual Studio开发环境就是利用一个进程编辑源文件，并利用另一个进程完成编译工作的应用程序。

线程（threading）：

• 线程有自己的一组CPU指令、寄存器与私有数据区，线程的数据可以与同一进程的线程共享。
• 当前的操作系统是面向线程的，即以线程为基本运行单位，并按线程分配CPU。

进程与线程有两个主要的不同点，其一是进程包含线程，线程使用进程的内存空间，当然线程也有自己的私有空间，但容量小；其二是进程有各自独立的内存空间，互不干扰，而线程是共享内存空间。
图三展示了进程、线程与CPU之间的关系。在图三中，进程一与进程二都含有3个线程，CPU会按照线程来分配任务，如图中4个CPU同时执行前4个线程，后两个标红线程处于等待状态，在CPU运行完当前线程时，等待的线程会被唤醒并进入CPU执行。通常，进程含有的线程数越多，则它占用CPU的时间会越长。

图三、进程、线程与CPU关系

3.2 全局解释器锁GIL：

GIL是计算机程序设计语言解释器用于同步线程的一种机制，它使得任何时刻仅有一个线程在执行。即便在多核心处理器上，使用 GIL 的解释器也只允许同一时间执行一个线程。Python的Cpython解释器（普遍使用的解释器）使用GIL，在一个Python解释器进程内可以执行多线程程序，但每次一个线程执行时就会获得全局解释器锁，使得别的线程只能等待，由于GIL几乎释放的同时就会被原线程马上获得，那些等待线程可能刚唤醒，所以经常造成线程不平衡享受CPU资源，此时多线程的效率比单线程还要低下。在python的官方文档里，它是这样解释GIL的：

In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)

可以说它的初衷是很好的，为了保证线程间的数据安全性；但是随着时代的发展，GIL却成为了python并行计算的最大障碍，但这个时候GIL已经遍布CPython的各个角落，修改它的工作量太大，特别是对这种开源性的语音来说。但幸好GIL只锁了线程，我们可以再新建解释器进程来实现并行，那这就是multiprocessing的工作了。

3.3 multiprocessing

multiprocessing是python里的多进程包，通过它，我们可以在python程序里建立多进程来执行任务，从而进行并行计算。官方文档如下所述：

The multiprocessing package offers both local and remote concurrency, effectively side-stepping the Global Interpreter Lock by using subprocesses instead of threads.
我们接下来介绍下multiprocessing的各个接口：

3.3.1 进程process

multiprocessing.Process(target=None,  args=())
    target: 可以被run()调用的函数，简单来说就是进程中运行的函数
    args: 是target的参数

process的方法：
    start(): 开始启动进程，在创建process之后执行
    join([timeout])：阻塞目前父进程，直到调用join方法的进程执行完或超时（timeout），才继续执行父进程
    terminate():终止进程，不论进程有没有执行完，尽量少用。

示例1

from multiprocessing import Process

def f(name):
    print \'hello\', name

if __name__ == \'__main__\':
    p = Process(target=f, args=(\'bob\',)) # p进程执行f函数，参数为\'bob\'，注意后面的“,”
    p.start() # 进程开始
    p.join() # 阻塞主线程，直至p进程执行结束

3.3.2 进程池Process Pools

class multiprocessing.Pool([processes])
    processes是进程池中的进程数，默认是本机的cpu数量
方法：
    apply(func[, args[, kwds]])进程池中的进程进行func函数操作，操作时会阻塞进程，直至生成结果。
    apply_async(func[, args[, kwds[, callback]]])与apply类似，但是不会阻塞进程
    map(func, iterable[, chunksize])进程池中的进程进行映射操作
    map_async(func, iterable[, chunksize[, callback]])
    imap(func, iterable[, chunksize])：返回有序迭代器
    imap_unordered(func, iterable[, chunsize])：返回无序迭代器
    close()：禁止进程池再接收任务
    terminate()：强行终止进程池，不论是否有任务在执行
    join()：在close()或terminate()之后进行，等待进程退出

示例2

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == \'__main__\':
    p = Pool(5) # 创建有5个进程的进程池
    print(p.map(f, [1, 2, 3])) # 将f函数的操作给进程池

3.3.3 Pipes & Queues

multiprocessing.Pipe([duplex])
    返回两个连接对象（conn1, conn2），两个连接对象分别访问pipe的头和尾，进行读写操作
    Duplex: True(default),创建的pipe是双向的，也即两端都可以进行读写；若为False，则pipe是单向的，仅可以在一端读，另一端写，此时与Queue类似。

multiprocessing.Queue([maxsize])
    qsize()：返回queue中member数量
    empty()：如果queue是空的，则返回true
    full()：如果queue中member数量达到maxsize，则返回true
    put(obj)：将一个 放入到queue中
    get()：从队列中取出一个 并将它从queue中移除，FIFO原则
    close()：关闭队列，并将缓存的 写入pipe

示例

from multiprocessing import Pool
import time
def f(x):
    return x*x
if __name__ == \'__main__\':
    pool = Pool(processes=4)              # start 4 worker processes
    result = pool.apply_async(f, (10,))   # evaluate \"f(10)\" asynchronously in a single process
    print result.get(timeout=1)           # prints \"100\" unless your computer is *very* slow
    print pool.map(f, range(10))          # prints \"[0, 1, 4,..., 81]\"
    it = pool.imap(f, range(10))
    print it.next()                       # prints \"0\"
    print it.next()                       # prints \"1\"
    print it.next(timeout=1)              # prints \"4\" unless your computer is *very* slow
    result = pool.apply_async(time.sleep, (10,))
    print result.get(timeout=1)           # raises multiprocessing.TimeoutError

3.3.4 进程锁multiprocessing.Lock

当一个进程获得（acquire）锁之后，其它进程在想获得锁就会被禁止，可以保护数据，进行同步处理。
     acquire(block=True, timeout=None)：尝试获取一个锁，如果block为true，则会在获得锁之后阻止其它进程再获取锁。
     release()：释放锁

3.3.5 共享内存——Value, Array

共享内存通常需要配合进程锁来处理，保证处理的顺序相同。

multiprocessing.Value(typecode_or_type, *args[, lock])
    返回一个ctype对象，
    创建c = Value(‘d’, 3.14),调用c.value()
multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)
    返回一个ctype数组，只能是一维的
    Array(‘i’, [1, 2, 3, 4])

3.3.6 其它方法

multiprocessing.active_children()：返回当前进程的所有子进程
multiprocessing.cpu_count()：返回本计算机的cpu数量
multiprocessing.current_process()：返回当前进程

3.3.7 注意事项：

1. 尽量避免共享数据
2. 所有对象都尽量是可以pickle的
3. 避免使用terminate强行终止进程，以造成不可预料的后果
4. 有队列的进程在终止前队列中的数据需要清空，join操作应放到queue清空后
5. 明确给子进程传递资源、参数

windows平台另需注意：

• 注意跨模块全局变量的使用，可能被各个进程修改造成结果不统一
• 主模块需要加上if name == 'main':来提高它的安全性，如果有交互界面，需要加上freeze_support()

四、multiprocessing实战

process、lock与value尝试：

import multiprocessing as mp
import time

def job(v, num, l):
    l.acquire() # 锁住
    for _ in range(5):
        time.sleep(0.1) 
        v.value += num # 获取共享内存
        print(v.value)
    l.release() # 释放


def multicore():
    l = mp.Lock() # 定义一个进程锁
    #l = 1
    v = mp.Value(\'i\', 0) # 定义共享内存
    p1 = mp.Process(target=job, args=(v,1,l)) # 需要将lock传入
    p2 = mp.Process(target=job, args=(v,3,l)) 
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__==\'__main__\':
    multicore()

上述代码即对共享内存叠加5次，p1进程每次叠加1，p2进程每次叠加3，为了避免p1与p2在运行时抢夺共享数据v，在进程执行时锁住了该进程，从而保证了执行的顺序。我测试了三个案例：

1. 直接运行上述代码输出[1, 2, 3, 4, 5, 8, 11, 14, 17, 20]，运行时间为1.037s
2. 在1的基础上注释掉锁（上述注释了三行），在没有锁的情况下，输出[1, 4, 5, 8, 9, 12, 13, 15, 14, 16],运行时间为0.53s
3. 在2的基础上将p1.join()调到p2.start()前面，输出为[1, 2, 3, 4, 5, 8, 11, 14, 17, 20]，运行时间为1.042s.

可以发现，没锁的情况下调整join可以取得与加锁类似的结果，这是因为join即是阻塞主进程，直至当前进程结束才回到主进程，若将p1.join（）放到p1.start()后面，则会马上阻塞主进程，使得p2要稍后才开始，这与锁的效果一样。
如果如上述代码所示，p1.join（）在p2.start()后面，虽然是p1先join()，但这时只是阻塞了主进程，而p2是兄弟进程，它已经开始了，p1就不能阻止它了，所以这时如果没锁的话p1与p2就是并行了，运行时间就是一半，但因为它们争抢共享变量，所以输出就变得不确定了。

pool

import multiprocessing as mp
#import pdb

def job(i):
    return i*i

def multicore():
    pool = mp.Pool()
    #pdb.set_trace()
    res = pool.map(job, range(10))
    print(res)
    res = pool.apply_async(job, (2,))
    # 用get获得结果
    print(res.get())
    # 迭代器，i=0时apply一次，i=1时apply一次等等
    multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
    # 从迭代器中取出
    print([res.get() for res in multi_res])

multicore()

pool其实非常好用，特别是map与apply_async。通过pool这个接口，我们只有指定可以并行的函数与函数参数列表，它就可以自动帮我们创建多进程池进行并行计算，真的不要太方便。pool特别适用于数据并行模型，假如是消息传递模型那还是建议自己通过process来创立进程吧。

总结

小子这次主要是按自己的理解把并行计算理了下，对进程、线程、CPU之间的关系做了下阐述，并把python的multiprocessing这个包拎了拎，个人感觉这个里面还大有学问，上次我一个师兄用python的process来控制单次迭代的运行时间（运行超时就跳过这次迭代，进入下一次迭代）也是让我涨了见识，后面还要多多学习啊。
感谢您花费宝贵的时间阅读到这里，希望能有所收获，也欢迎在评论区进行交流。

推荐好文：
multiprocessing官方文档
python多进程的理解 multiprocessing Process join run(推荐好文)
多进程 Multiprocessing