Python 多进程编程
2018年7月9日
编程
# Python 多进程对效率的提升 一篇[《Python 中单线程、多线程和多进程的效率对比实验》][multiprocess-efficiency]的文章中提到: > Python是运行在解释器中的语言,有一个全局锁(GIL), > 在使用多进程(Thread)的情况下,不能发挥多核的优势。 > 而使用多进程(Multiprocess),则可以发挥多核的优势真正地提高效率。 文章中对 Python 在多线程、多进程的效率进行了对比: | 操作类型 | CPU 密集型 | IO 密集型 | 网络请求密集型 | | ---------- | -------------: | -------------: | -------------: | | 线性操作 | 94.91824996469 | 22.46199995279 | 7.3296000004 | | 多线程操作 | 101.1700000762 | 24.8605000973 | 0.5053332647 | | 多进程操作 | 53.8899999857 | 12.7840000391 | 0.5045000315 | 可见: 1. 多线程操作只在网络请求密集型操作中具有非常明显的优势,其开销小于多进程,可用于网络爬虫。 2. 多进程操作在各种操作中都有效率提升,在 IO 密集型操作中的优势更大。 最近在处理一套出租车数据,出租车数据量非常大,自己搭建数据库, 查询效率非常低。因此采用 Python 脚本进行处理, # Python 开启多进程 Python 中的 `multiprocess` 包提供了多进程支持。可以使用三种方法来创建进程。 ## 创建 Process 对象 最简单的开启 Python 进程的方法,是直接构造 `multiprocess.Process` 对象 ```python from multiprocess import Process process = Process() ``` ### 构造参数 `Process` 对象在构造时主要接收三个参数: - `target`:进程调用的函数; - `args`:进程调用函数时给函数传递的参数,为一个元组; - `name`:别名。 ### 属性 `Process` 的类型有以下属性: - `daemon`:当这个属性设置为 `True` 时,子进程会随着主进程的结束而结束。否则,主进程结束后,子进程依然会继续进行; - `exitcode`:进程在运行时为 None 、如果为 –N ,表示被信号 N 结束; - `name` - `pid` - `authkey` ### 方法 `Process` 有如下方法: - `start()`:调用 `start()` 函数时,子进程开始执行,主进程继续执行。 - `join()`:“[阻塞当前进程,直到调用 join 方法的那个进程执行完,再继续执行当前进程。][join-explain]” - `run()`:当构造时如果没有制定 `target` 参数,那么 `start()` 方法默认执行 `run()` 函数。 - `is_alive()`:判断当前进程是否活动。 ### 调用示例 ```python from multiprocess import Process import os import math def work_fun(work_list): pass def distrib_works(work_list, process_num): group_length = math.ceil(len(filename_list) / process_num) return [work_list[(i*group_length):((i+1)*group_length)] for i in range(process_num)] work_list = os.listdir("../data") process_num = 4 group = distrib_works(work_list, process_num) process_list = [Process(target=work_fun, args=(g,)) for g in group_list] for p in process_list: p.daemon = True p.start() for p in process_list: p.join() ``` ## 将进程定义为类 利用 Python 面向对象的特性,我们可以创建一个类,继承 `Process` 类, 将一些数据直接在构造的时候保存下来,可以无需在调用的时候传入。 例如,当我们在多进程程序中使用 `tqdm` 库显示进度条时,会用到其 `position` 参数来指定当前进度条在控制台中显示的位置,这个参数的值,我们可以直接保存在进程类中, 无需调用的时候再传入。 将进程定义为类的方法如下: ```python from multiprocess import Process class MyProcess(Process): position = 0 works = None def __init__(self, position, works) Process.__init__(self) self.position = poisition self.works = works def run(): pass ``` 如果在构造函数中,调用的 `Process` 的构造函数没有指定 `target`, 进程同样默认执行 ***不带参数的 `run` 函数,即使你的 `run` 函数定义了形参!*** 在创建进程时,只需要将原来调用的 `Process` 的构造函数,改为调用 `MyProcess` 的构造函数即可。 这种创建进程方式的实例如下: ```python from multiprocess import Process, Lock def distrib_works(work_list: List[str], process_num) -> List[List[str]]: group_length = math.ceil(len(work_list) / process_num) return [g for g in [work_list[(i*group_length):((i+1)*group_length)] for i in range(process_num)] if len(g) > 0] class FindTargetTaxiProcess(multiprocessing.Process): def __init__(self, input_files, index, lock, log_file): multiprocessing.Process.__init__(self, target=find_target, args=(lock, log_file)) self.input_files = input_files self.index = index def find_target(self, lock, log_file): for filename in self.input_files: with open(filename) as in_file: for row in tqdm(in_file, ncols=80, position=self.index): cells = row.split(",") if int(cells[0]) == 11865: print(row) if __name__ == '__main__': LOG_FILE = r"E:\出租车点\上下车点\scripts\data\find_error.log" lock = multiprocessing.Lock() # ROOT_DIR = "../data/201502/temp" ROOT_DIR = r"E:\出租车点\201502\RawCSV" INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)] GROUP_LIST = distrib_works(INPUT_FILES, 4) PROCESS_LIST = [FineTargetTaxiProcess(element, i, lock) for i, element in enumerate(GROUP_LIST)] for process in PROCESS_LIST: process.start() ``` > 这种定义为类的方式有一个好处,在用 VSCode 调试的时候,在子进程中打断点是无效的。 > 如果用这种方式,可以将调用的 `start()` 函数改为 `run()` 或其他实际进程执行的函数, > 这样就可以调试进程内部了。当解决了 Bug 后,就可以换回 `start()` 函数并行执行。 ## 进程池(Pool) 可以发现,上面两种创建进程的方式,都是用到了一个 `distrib_works()` 函数来分配各个进程的任务。 这一过程可以被一个叫做进程池的类型代替。 ### 进程池构造 构造进程池的方法非常简单,导入 `Pool` 之后,直接构造 `Pool` 对象。 构造时可以指定最多的进程数量,默认是 CPU 核心数。 ```python from multiprocess import Pool p = Pool(process=6) ``` ### 进程池方法 `Pool` 类型主要有以下方法: - `apply_async()` 和 `apply()`:这两个函数都是让进程池开始执行任务,`apply_async()` 是非阻塞的(主进程继续执行),`apply()` 是阻塞的(主进程等待子进程执行完成后继续执行)。 - `close()`:关闭进程池,不再接收新任务。 - `join()`:[主进程阻塞,等待子进程的退出, join方法要在close或terminate之后使用。][python多进程-cnblogs] ### 进程池调用示例 将上面一段使用 `FindTargetTaxiProcess` 类编写的代码用 `Pool` 重写: ```python from multiprocess import Process, Pool, Lock def find_target(in_file, lock, log_file): with open(filename) as in_file: for row in tqdm(in_file, ncols=80, position=self.index): cells = row.split(",") if int(cells[0]) == 11865: with lock: with open(log_file, mode="a") as log: print(row, file=log) if __name__ == '__main__': LOCK = multiprocessing.Lock() LOG_FILE = r"E:\出租车点\上下车点\scripts\data\find_error.log" ROOT_DIR = r"E:\出租车点\201502\RawCSV" INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)] POOL = Pool(process=6) for f in INPUT_FILES: POOL.apply_async(find_target, (f, LOCK, LOG_FILE)) POOL.close() POOL.join() ``` # 多进程共享资源 当我们把多个任务分解到 $n$ 个进程上执行时,这 $n$ 个进程往往会存在某种共享的资源, 如共享一个控制台、文件系统、列表或字典。这里存在两个问题: - 当多个进程同时访问这些资源时,就会产生冲突。例如,两个进程同时对控制台输出文本,写入的结果可以错综复杂,并不是两段文本的顺序组合。 - 各个进程有自己的内存空间,变量无法共享。例如,当想要利用多个进程操作主进程的一个列表时,各个进程操作结束后,主进程仍然是原来的状态。 这两个问题的解决,前者靠“锁”机制,后者靠“共享变量”机制。 ## 锁 对于冲突的情况,当使用 tqdm 显示多个进度条时比较明显。在 Windows 上,由于 “tqdm 无法获取默认锁”,因此控制台输出会比较乱,下面是一段程序在 Windows 上运行的效果: ```plain λ python3 find_errors.py Process 0: 0it [00:00, ?it/s] Process 1: 0it [00:00, ?it/s] Process 0: 273516it [00:00, 523719.79it/s] Process 0: 995883it [00:01, 510379.67it/s] Process 0: 1107387it [00:02, 510326.10it/s] Process 0: 1224813it [00:02, 512761.81it/s] Process 0: 3483799it [00:06, 539191.83it/s] Process 1: 3683852it [00:06, 571536.15it/s] Process 0: 3550015it [00:06, 540296.03it/s] Process 0: 3615558it [00:06, 540947.45it/s] Process 0: 3742521it [00:06, 542112.37it/s] ``` 而在 Linux 系统中的运行结果是 ```plain Process 0: 2045720it [00:03, 647073.52it/s] Process 1: 2092184it [00:03, 661530.01it/s] Process 2: 2065411it [00:03, 652446.31it/s] Process 3: 2093610it [00:03, 661782.04it/s] ``` 可见在访问共享资源的时候,加锁是非常有必要的。 ### 互斥锁(Lock) `Lock` 属于“互斥锁”,即[保证在任一时刻,只能有一个线程访问该对象。][使用Lock互斥锁] 通过 `Lock` 类型创建互斥锁后,将其传递到子进程内部,即可在子进程中使用。 使用 `Lock` 时,可以使用 `with` 语句加锁, `with` 语句块执行完成后自动解锁; 也可以通过其 `acquire()` 函数来加锁,使用 `release()` 函数解锁。 使用 `with` 语句进行加锁的示例代码如下: ```python def run(self): for filename in self.input_files: with open(filename, encoding="GB2312") as in_file: for row in tqdm(in_file): cells = row.split(",") if int(cells[0]) == 11865: with self.lock: with open(TARGET_TAXI_FILE, mode="a") as log: print(row, file=log) ``` > 这段代码在 Windows 上运行时,子进程内部的 lock 和 主进程传递进去的 lock 的 id 值不相同。 > 但是在 Linux 系统上时相同的。因此 Windows 上这段代码有可能会出错。 > 不过当文件被一个进程打开时,是无法被另一个进程打开的,因此这段程序的结果倒没出什么错。 ### 可重入锁(RLock) 互斥锁可以解决简单的避免资源冲突的问题,但当一个线程加锁后仍需要再次访问共享资源时, 就形成了嵌套锁,而使用互斥锁时就形成了“死锁”问题。这时我们需要使用 `RLock` 类型, 即“可重入锁”。 > 死锁的含义是:是指两个或两个以上的进程在执行过程中,因争夺资源而造成的一种互相等待的现象, > 若无外力作用,它们都将无法推进下去。 > 此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程。 > 避免死锁主要方法是正确有序的分配资源。 `multiprocess` 中的 `RLock` 类型与 `Lock` 类型的区别在于: [RLock允许在同一线程中被多次申请。而 Lock 却不允许这种情况。][Python中Lock与RLock] 因此,[如果使用 `RLock` ,那么 acquire() 和 release() 必须成对出现][Python中Lock与RLock], 调用了几次 `acquire()`,就需要调用几次 `release()`。 ### 条件锁(Condition) 条件同步机制是指:线程 $B$ 等待特定条件 $C$ ,而另一个线程 $A$ 发出特定条件满足的信号 $C$ 。 $B$ 在收到信号 $C$ 时,继续执行。 可以通过“生产者-消费者”模型来理解这一过程。 生产者获取锁,生产一个随机整数,通知消费者并释放锁。 消费者获取锁,如果有整数则消耗一个整数并释放锁,如果没有就等待生产者继续生产。 示例代码如下(参考[《Python 线程同步机制》][Python线程同步机制]并进行修改): ```python import multiprocess class Producer(multiprocessing.Process): def __init__(self, productList, condition): multiprocessing.Process.__init__(self) self.productList = productList # type: List self.condition = condition # type: multiprocess.Condition def run(self): while True: product = random.randint(0, 100) with self.condition: print("条件锁:被 生产者 获取") self.productList.append(product) print(f"生产者:产生了 {product}。") print("生产者:唤醒消费者线程") self.condition.notify() print("条件锁:被 生产者 释放") time.sleep(1) class Customer(multiprocessing.Process): def __init__(self, productList, condition): multiprocessing.Process.__init__(self) self.productList = productList # type: List self.condition = condition # type: multiprocess.Condition def run(self): while True: with self.condition: print("条件锁:被 消费者 获取") while True: if self.productList: product = self.productList.pop() print(f"消费者:消费了 {product}") break print("消费者:等待生产者") self.condition.wait() print("条件锁:被 消费者 释放") def main(): manager = multiprocessing.Manager() productList = manager.list() condition = multiprocessing.Condition() process_producer = Producer(productList, condition) process_customer = Customer(productList, condition) process_producer.start() process_customer.start() process_producer.join() process_customer.join() if __name__ == '__main__': main() ``` 运行的部分结果是: ```plain 条件锁:被 生产者 获取 生产者:产生了 47。 生产者:唤醒消费者线程 条件锁:被 生产者 释放 条件锁:被 消费者 获取 消费者:消费了 47 条件锁:被 消费者 释放 条件锁:被 消费者 获取 消费者:等待生产者 条件锁:被 生产者 获取 生产者:产生了 100。 生产者:唤醒消费者线程 条件锁:被 生产者 释放 消费者:消费了 100 条件锁:被 消费者 释放 条件锁:被 消费者 获取 消费者:等待生产者 条件锁:被 生产者 获取 生产者:产生了 95。 生产者:唤醒消费者线程 条件锁:被 生产者 释放 消费者:消费了 95 条件锁:被 消费者 释放 条件锁:被 消费者 获取 消费者:等待生产者 ``` ### 信号量(Semaphore) 信号量是一个非负整数,所有通过它的进程都会将该整数减一, 当该整数值为零时,所有试图通过它的进程都将处于等待状态。 ```python from multiprocessing import Process, current_process, Semaphore import time def worker(s, i): s.acquire() print(current_process().name + "acquire"); time.sleep(i) print(current_process().name + "release\n"); s.release() if __name__ == "__main__": s = Semaphore(2) for i in range(5): p = Process(target = worker, args=(s, i*2)) p.start() ``` ## 共享变量 在多进程中,是无法直接使用全局变量作为共享变量的,因为不同进程具有不同的内存空间。 但是,共享变量也是不能避免的。Python 中也提供了一些创建共享变量的方法。 - Multiprocess 包内置类型 - 通过 Manager 创建共享变量 ### multiprocess 包内置类型 multiprocess 包提供了两种类型的共享变量: - `Value(typecode_or_type, *args, lock=True)`:表示一个值类型变量。 - `Array(typecode_or_type, size_or_initializer, *, lock=True)`:表示一个数组。这种创建数组的方式能力比较有限,它不支持除了 C 数据类型以外的类型。 `typecode_or_type` 描述了元素的类型,可取值是: | typecode | type | | -------- | --------------- | | 'c' | ctypes.c_char | | 'u' | ctypes.c_wchar | | 'b' | ctypes.c_byte | | 'B' | ctypes.c_ubyte | | 'h' | ctypes.c_short | | 'H' | ctypes.c_ushort | | 'i' | ctypes.c_int | | 'I' | ctypes.c_uint | | 'l' | ctypes.c_long | | 'L' | ctypes.c_ulong | | 'f' | ctypes.c_float | | 'd' | ctypes.c_doubl | 创建后,只要将这些变量传递给子进程即可。 ### 通过 Manager 创建共享变量 Manager() 返回的 manager 对象提供一个服务进程,使得其他进程可以通过代理的方式操作 Python 对象。 Manager 支持 list、dict 等多种数据类型。 ([多进程multiprocess][liujiang]) 把之前的共享变量的代码中,共享的变量由 list 改为 Manager 对象创建的 list,可以得到正确结果。 ```python from multiprocessing import Process, Lock, Manager import time def work(lock, var, index): with lock: var.append(index) print(f"Process {index} apped {index}") if __name__ == '__main__': var = Manager().list() lock = Lock() process_list = [Process(target=work, args=(lock, var, i)) for i in range(8)] for p in process_list: p.start() for p in process_list: p.join() print(var) ``` # 进程间通信 进程间通信,可以起到共享变量的效果,也可以起到锁的效果。 进程间通信的方式有三种: - 事件(Event) - 队列(Queue) - 管道(Pipe) ## 通过事件(Event)通信 Event 是同步通信的方式,有些类似于条件锁。由于是它是同步的,而且不能传递数据。 因此这里就不仔细研究 Event 的作用。 这个例子示例了主进程与子进程之间通过 Event 进行通信的方法。 ```python import multiprocessing import time def wait_for_event(e): print("wait_for_event: starting") e.wait() print("wairt_for_event: e.is_set()->" + str(e.is_set())) def wait_for_event_timeout(e, t): print("wait_for_event_timeout:starting") e.wait(t) print("wait_for_event_timeout:e.is_set->" + str(e.is_set())) if __name__ == "__main__": e = multiprocessing.Event() w1 = multiprocessing.Process(target=wait_for_event, args=(e,)) w2 = multiprocessing.Process(target=wait_for_event_timeout, args=(e, 6)) w1.start() w2.start() time.sleep(10) print("main: event setting") e.set() print("main: event is set") ``` ## 通过队列(Queue)通信 Queue 是多进程安全的队列,可以使用 Queue 实现多进程之间的数据传递。 Queue 有两个方法: - `put()`:将数据插入队列中。 - `get()`:从队列读取并且删除一个元素。 这两个方法都有两个参数:`blocked`, `timeout`, 控制队满和队空两种情况: - `put`:当队满时,如果 `blocked=True` ,那么会阻塞 `timeout` 指定的时间,直到队列有空间。如果超时,或 `blocked=False` ,则抛出 `Queue.Full` 异常。 - `get`:当队满时,如果 `blocked=True` ,那么会阻塞 `timeout` 指定的时间,直到队列有元素。如果超时,或 `blocked=False` ,则抛出 `Queue.Empty` 异常。 调用实例: ```python class FineTargetTaxiProcess(mp.Process): ''' 处理进程:多进程方式处理文件,结果全部传递给打印进程。 ''' def __init__(self, input_files, index, queue): mp.Process.__init__(self, target=self.pick, args=(queue,)) self.input_files = input_files self.index = index self.lock = lock def pick(self, queue): for filename in tqdm(self.input_files, ncols=80, position=self.index, desc=f"Process {self.index}"): with open(filename, encoding="GB2312") as in_file: for row in in_file: cells = row.split(",") if int(cells[0]) == 11865: try: queue.put(",".join(cells), block=False) except: print("Queue full") class PrinterProcess(mp.Process): ''' 打印进程:维持对输出文件的打开状态,打印数据。 可避免频繁打开、关闭结果文件造成的系统开销, 但是引入了消息传递的开销。 ''' def __init__(self, output_file, log, queue): mp.Process.__init__(self, target=self.write, args=(queue,)) self.output_file = output_file self.log_file = log def write(self, queue): with open(self.output_file, mode="w", newline="\n") as printer, open(self.log_file, mode="w") as log: while True: try: row = queue.get(block=True, timeout=1) print(row, file=printer) except: print("Queue empty", file=log) if __name__ == '__main__': lock = mp.Lock() ROOT_DIR = r"/mnt/e/出租车点/201502/RawCSV" INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)] GROUP_LIST = distrib_works(INPUT_FILES, 6) QUEUE = mp.Queue() PROCESS_LIST = [FineTargetTaxiProcess(element, i, QUEUE) for i, element in enumerate(GROUP_LIST)] PRINTER_PROCESS = PrinterProcess("./data/usequeue.txt", "./data/usequeue.log", QUEUE) for process in PROCESS_LIST: process.daemon = True process.start() PRINTER_PROCESS.daemon = True PRINTER_PROCESS.start() for p in PROCESS_LIST: p.join() ``` ## 通过管道(Pipe)通信 Pipe 是一个可以双向通信的对象,返回 `(conn1, conn2)`, 代表一个管道的两个端, `conn1` 只负责接受消息, `conn2` 只负责发送消息。 如果设置了 `duplex=True` ,那么这个管道是全双工模式, `conn1` 和 `conn2` 均可收发。 ```python class FineTargetTaxiProcess(mp.Process): ''' 处理进程:多进程方式处理文件,结果全部传递给打印进程。 ''' def __init__(self, input_files, index, pipe): mp.Process.__init__(self, target=self.pick, args=(pipe,)) self.input_files = input_files self.index = index self.lock = lock def pick(self, pipe): for filename in tqdm(self.input_files, ncols=80, position=self.index, desc=f"Process {self.index}"): with open(filename, encoding="GB2312") as in_file: for row in in_file: cells = row.split(",") if int(cells[0]) == 11865: try: pipe.send(",".join(cells)) except e as Exception: print("Pipe send error") class PrinterProcess(mp.Process): ''' 打印进程:维持对输出文件的打开状态,打印数据。 可避免频繁打开、关闭结果文件造成的系统开销, 但是引入了消息传递的开销。 ''' def __init__(self, output_file, log, pipe): mp.Process.__init__(self, target=self.write, args=(pipe,)) self.output_file = output_file self.log_file = log def write(self, pipe): with open(self.output_file, mode="w", newline="\n") as printer, open(self.log_file, mode="w") as log: while True: try: row = pipe.recv() print(row, file=printer) except e as Exception: print("Pipe read error", file=log) if __name__ == '__main__': lock = mp.Lock() # ROOT_DIR = "../data/201502/temp" ROOT_DIR = r"/mnt/e/出租车点/201502/RawCSV" INPUT_FILES = [os.path.join(ROOT_DIR, f) for f in os.listdir(ROOT_DIR)] GROUP_LIST = distrib_works(INPUT_FILES, 6) (RECEIVER, SENDER) = mp.Pipe() PROCESS_LIST = [FineTargetTaxiProcess(element, i, SENDER) for i, element in enumerate(GROUP_LIST)] PRINTER_PROCESS = PrinterProcess("./data/usepipe.txt", "./data/usepipe.log", RECEIVER) for process in PROCESS_LIST: process.daemon = True process.start() PRINTER_PROCESS.daemon = True PRINTER_PROCESS.start() for p in PROCESS_LIST: p.join() ``` # 其他 ## tqdm 多进度条 是一个快速,可扩展的 Python 进度条,可以在 Python 长循环中添加一个进度提示信息, 用户只需要封装任意的迭代器 `tqdm(iterator)` 。 这里有一些参数: - `ncols`:整个进度条(包括条以及其他文字)的宽度。最好设置一个小于控制台总宽的值。 - `mininterval`:进度条更新的最小间隔。默认为 0.1。 - `position`:进度条的位置,从0开始。对不同的 tqdm 对象设置不同的 position,可以在控制台的不同位置显示出来,适用于多进程与多线程。 > 由于 Windows 上多进程时 tqdm 无法获取默认的锁,所以会出现进度条错乱。在 Linux 上是没有问题的。 ## Windows 上 Lock 的问题 其实每次传入子进程函数内部的 `Lock`,在各个进程中的 `id` 都不一样。在 Linux 下没有这个问题。 这往往会导致一些程序在 Windows 上不正确。 因此,在 Windows 上最好少用 `Lock`,多采用消息传递或共享变量的方式设计程序。 [multiprocess-efficiency]: https://segmentfault.com/a/1190000007495352 [join-explain]: https://www.cnblogs.com/lipijin/p/3709903.html [python多进程-cnblogs]: http://www.cnblogs.com/kaituorensheng/p/4445418.html [Python中Lock与RLock]: https://blog.csdn.net/cnmilan/article/details/8849895 [使用Lock互斥锁]: https://www.jb51.net/article/63508.htm [Python线程同步机制]: https://yoyzhou.github.io/blog/2013/02/28/python-threads-synchronization-locks/ [liujiang]: http://www.liujiangblog.com/course/python/82
感谢您的阅读。本网站 MyZone 对本文保留所有权利。