并行编程：进程与线程

进程

进程就是一个执行中程序的实例。

对于Python的编程以及程序，我们通常要经历很久的等待时间，而这种等待的原因分为两个：

$I/O$限制

主要是由于I/O设备的处理速度跟不上CPU处理速度造成的等待。
$CPU$限制

在进行很大数据或者复杂程度比较高的运算的时候经常遇到的CPU速度的限制。

而要解决这两种限制，我们必须要了解进程以及并发。

首先要知道，系统中所有的程序都运行在某一个进程的上下文当中。而上下文就是程序正确运行所需要的状态。可以包括程序的代码和数据、栈以及寄存器内容、$PC$、环境变量、打开文件描述符集合等等，你并不需要都知道这是什么东西，你只要知道这是程序正确运行的状态就可以了。

当你的程序在你的电脑上运行的时候，你会想当然地以为整个系统只有这个程序在运行，那是因为操作系统给你提供了一种假象，让你这么以为。那么我们也可以这么说：

进程是操作系统对一个正在运行的程序的一种抽象。

我们当然知道编译器的运行其实就相当于对shell说“运行这个程序”，那么这个时候我们的shell做了什么呢？此时，shell会创建一个新的进程，然后在这个进程的上下文当中运行你的文件，也就好像我们的程序占据了整个系统。

而这种假象大致可以分为两点：

独立逻辑控制流的假象：好像我们的程序在独占处理器？
私有地址空间的假象：好像我们的程序在独占内存系统？

下面我们一个个解释。

逻辑控制流

不要被名字吓到，这个玩意其实相当简单。现在你的系统中有许多的程序在运行，但是我们不要忘了：

在任何一个时刻，单处理器系统都只能处理一个进程的代码。

于是我们有了一种流，表示进程运行的序列，也可以简称为逻辑流。

LCS

上面的这个图可以很好的说明这个事了吧。首先$A$进程运行了一个时间片，然后操作系统内核进行了上下文切换，开始进行进程$B$，因为进程$B$只有一个时间片，所以直接进行完了，接着又是上下文切换到了进程$C$，运行完一个时间片之后切换完成了$A$，然后反过头来又进行完了$C$。

其实整个步骤的关键就在于进程是轮流使用控制器的，而逻辑控制流就是这么一个东西。

上下文切换

关于上面的上下文切换我们先简单介绍一下，这个东西是操作系统转换执行进程的手段，是一种较高层次形式的一场控制流，主要由操作系统内核管理，步骤大致如下：

保存当前进程的上下文
回复新进程的上下文
传递控制权

其实一个进程并不一定能够从头到尾直接一条龙的执行完毕，在进程执行的时候，内核可以决定抢占这个进程，然后重新开始另外一个进程——当然它之前也被抢占过。而这种决策也叫作调度，由内核中的调度器执行。

而这个玩意的意义在哪里呢？因为我们使用系统的过程可能并不是一帆风顺的，往往当前的进程会由一些$unexpected$的事故导致阻塞，那么这时内核的调度器就可以选择休眠这个进程，进行别的进程。你也可以很简单的使用你的C++来实现这个休眠，一个sleep就可以了。下图展示了之前的举例的逻辑控制流的$A$到$B$的切换。

Change

私有地址空间

地址空间表示任何一个计算机实体所占用的内存大小。比如外设、文件、服务器或者一个网络计算机。地址空间包括物理空间以及虚拟空间。

在一台地址为$n$位的机器上，地址空间表示$2^n$个可能地址的集合，而进程为每个程序提供私有地址空间。在计算机中，每一个设备以及进程都被分配了一个地址空间。

为什么说是私有呢？因为一般情况下一个空间中与某个地址相关联的内存字节是不能该被其他进程读写的。下面给出的是这个空间的通用结构：

关于地址空间的介绍不多说了，想要详细了解的请自行维基。

线程

这个玩意可以视作是一个“单位”，一个进程的单位。因为在现代的系统当中，每一个进程都是有多个称为线程的执行单元组成的，每一个线程都运行在进程的上下文当中，共享代码和全局数据。

一条线程指的是中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。

我们的Python支持的不仅仅是多进程，也支持多线程，其实这两个玩意的最终目的都是同时执行多个编程任务。在计算机科学当中，运行在操作系统中的每一个进程可以拥有多个线程，每一个进程都有自己独享的内存，也就是说这个进程里面的多个线程可以共享一块内存，但是多个进程之间就不行了，它们必须进行显式的通信。因此，可以认为多线程之间比多进程之间更容易共享数据，使得程序运行的更快，因此，由于网络服务器中对并行处理的需求，线程正在成为越来越重要的编程模型。

并发

方才我们提到了逻辑控制流，而计算机系统中有非常多类型的逻辑流，其中就有一种叫做“并发流”。它的定义是：一个逻辑流的执行在时间上与另一个流重叠。但是不要混淆概念：是重叠，但并不代表是同时执行。就好像最开始举得例子：A与B是并发，但是B和C就不是，因为B和C并没有重叠部分。

多个流并发地执行的一般现象被称为并发。一个进程和其他进程轮流运行的概念称为多任务。

以上的内容当然十分简单，但是下面我们要做的是对进程、线程的代码方面的实际应用。

进程池实现

这里我们的进程池的实现方法是多进程，大家必然不陌生这个东西。现在我们尝试创建两个进程，让每一个进程都输出一个”WORKING”，最后让主进程输出”WORKOUT!!!”结束。

import multiprocessing as MP, time, os

def PRINT(name, MSG):
	time.sleep(2)
	print("process %s %s says: %s" % (name, str(os.getpid()), MSG))

if __name__ == '__main__':
	print("process %s says: %s" % (os.getpid(), "WORKING"))	

	task1 = MP.Process(target = PRINT, args = ("1", "WORKING"))
	print("Now processing!")
	task1.start()

	task2 = MP.Process(target = PRINT, args = ("1", "WORKING"))
	print("Now processing!")
	task2.start()

	task1.join()
	task2.join()

	print('Process close')

我们从上到下解释一下。首先我们目前用的多进程模块就叫做”multiprocessing”，其中有一个类叫做”Process”，我们可以用这个类完成一个进程对象的描述。

1	if __name__ == '__main__'

这一段的意思我还是解释一下，如果你并非新手当然可以跳过。

我们知道Python的代码有两种使用方式，一种是直接运行，另一种就是import进另一份代码里面。也就是说一种是脚本执行，另一种是模块调用。而在上面这行代码下的代码只有在脚本执行的时候才会被执行。为什么呢？我们每一份代码（也称作模块）都有一个内置的变量$__name__$，当模块被直接执行的时候，$__name__$就等于文件名，而如果import到其他模块当中就等于那个模块的名称。而$__main__$一直等于自己模块的名称，因此只有脚本执行的时候才会有上述的语句为真。

然后我们看到对于task1和task2的定义是这样的。

1	task1 = MP.Process(target = PRINT, args = ("1", "WORKING"))

其中target是执行函数的名字，args是函数需要的参数元组。

然后start方法启动进程，join方法实现进程同步，等待所有进程退出。

在上面的代码中我们看到每一个进程都是输出了一遍WORKING，而进程之间是不能够共享数据的。

下面介绍的东西叫做Pool，也是multiprocessing模块里面的内容，作用是预先构造进程池，然后每次使用进程的时候会在进程池里面申请一个进程。对于apply方法，可以理解为“进程排队执行”，每一个进程都是从头进行到尾，也就是只有一个时间片。

from multiprocessing import Process, Pool, Manager
import time, os

def PRINT(MSG):
	time.sleep(1)
	print(MSG, os.getpid())

if __name__ == '__main__':
	pool = Pool(10)

	for i in range(10):
		pool.apply(func = PRINT, args = (str(i) + "said 'Hello'",))
		print(i, "Prepare OK")

	pool.close()
	pool.join()

close顾名思义，在调用它之后们不会有任何新的进程加入到pool。

另外还有一个apply_async模块支持回调。

from multiprocessing import Process, Pool, Manager
import time, os

def PLUS(MSG):
	time.sleep(1)
	return MSG + 100

def PRINT(MSG):
	print(MSG)

if __name__ == '__main__':
	pool = Pool(5)
	for i in range(10):
		pool.apply_async(func = PLUS, args = (i,), callback = PRINT)
		print(i, "Prepare OK")
	pool.close()
	pool.join()

以上这个程序会输出100 ~ 109的所有数。

线程实现

Python内部也提供了很好的模块叫做threading，我们可以用它来实现线程。

import threading

def Doit(MSG):
	PRINT(MSG)

def PRINT(MSG):
	print(threading.current_thread(), "says:", MSG)

if __name__ == '__main__':
	PRINT("I'm main")
	for i in range(5):
		thread = threading.Thread(target = Doit, args = ("I'm child",))
		thread.start()

threading.current_thread()是返回当前的Thread对象，对应于调用者的控制线程。start()也是启动线程活动。这里没有加close()、join()之类的函数是因为默认情况下程序就会等待线程全部执行完毕才停止的，但是也可以更改为后台线程，让主线程优先。

实例分析：文章编写与上传

假设你是一个网络文章工作者（我也不知道这个名称是怎么来的），被分配了两种任务：编写文章到本地以及将本地文章上传到云上。这都是很简单的工作以至于你一个人就可以完成，那么你的工作流程大致可以分为两种：

每次编写完一篇文章都即时上传，然后删除本地的这篇文章。
先编写完所有的文章之后统一上传，然后删除所有的文章。

这两种方法看起来好像没有很大的区别，但是第二种方法却要临时地占用更大的本地磁盘空间，相比之下就不如第一步来的优。

那么如果这个工作被分配了另外一个人呢？那么工作的分配可以是一个人负责编写文章，另一个人负责上传，这都需要一定的时间，那么为了工作的流畅性，就要考虑两个人速度的问题，因为如果编写文章的人的速度非常快，那么就有可能出现文章堆积的情况。要么在上传文章的人上传完它那篇文章之前一直闲着，要么就把它堆积下来，而总的来说处理的最终速度都是取决于最慢的那个人。

multiprocessing模块里面有一个queue类，也就是队列，想必大家肯定都不陌生，我也就不再解释。

import multiprocessing, time

def writer(articles, queue):
	for article in articles:
		time.sleep(1)
		print('Writing', article)
		queue.put(article)

def updater(queue):
	while True:
		time.sleep(1)
		article = queue.get()
		print('Updating', article)
		queue.task_done()

if __name__ == '__main__':
	Queue = multiprocessing.JoinableQueue()
	Updater = multiprocessing.Process(target = updater, args = (Queue,))
	Updater.daemon = True
	Updater.start()

	Articles = ['A', 'B', 'C', 'D']
	writer(Articles, Queue)

$queue.task_done()$：使用者使用此方法发出信号，表示queue.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常。

$queue.join()$：生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用queue.task_done（）方法为止。

$p.daemon=True$：设置为守护进程，在主线程停止时p也停止，但是不用担心，producer内调用q.join保证了consumer已经处理完队列中的所有元素