python多线程多进程concurrent.futures详细总结

基本代码

import concurrent.futures

def f(i):
    print(i)
    return i

if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(f, i) for i in range(10)]

        for future in concurrent.futures.as_completed(futures):
            print(future.result())
            
            if future.exception():
                print(future.exception())

1. 代码解释

通过以上方式就可以多个进程同时运行f函数, future.result() 是函数返回的i, max_worker 是进程/线程数, 默认为 CPU 核心数, f 为要执行的函数, i 为传给 f 的参数, 其中 ProcessPoolExecutor 多进程可以替换为 ThreadPoolExecutor 多线程, 通过查资料发现:

ProcessPool is for CPU bound tasks so you can benefit from multiple CPU.
Threads is for io bound tasks so you can benefit from io wait.

如果是 `IO` 密集多尽量用多线程, `CPU` 密集尽量用多进程.

2. 异常处理

运用多进程/多线程的时候函数报错并不一定会直接结束程序, 而有可能会什么都不发生, 这样需要在 as_completed 之后捕捉异常, 用上面代码所示语句就可以.

3. `submit` 和 `map` 的区别

上面是 submit 函数, 还有另一种 map 函数的用法

import concurrent.futures

def f(i):
    print(i)
    return i

if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        futures = executor.map(f, range(10))
        
        # 等执行完全部返回
        print(futures)
        
        for future in futures:
            print(future)

map 函数会比 submit 更简洁, 但是没法对返回一个一个处理, 而 submit 的好处是可以对每个返回在执行完成的瞬间处理, 不用等到每个线程/进程都执行完毕.

4. 注意

我自己用过很多次这个模块, 也遇到过很多坑, 以下是一些需要注意的地方

该模块一定要写在 if __name__ == '__main__' 之后, 否则很容易报错

不能同时操作一个 class , 会报错, 要想操作类都是每个进程分别操作新的类, 如: futures = [executor.submit(Class().fun(), i) for i in range(10)]

5. 工程代码示例

    def get_data(self, workers=5):
        self.cookie = self.get_cookie(self)

        total_data = {
            'data'       : [],
            'status_code': 0
        }

        with concurrent.futures.ProcessPoolExecutor(max_workers=workers) as executor:
            futures = [executor.submit(self.analyze, patent_number) for patent_number in self.patent_number_list]
            for future in concurrent.futures.as_completed(futures):
                # add result to total data
                total_data['data'].append(future.result())

        return total_data

以上代码是在每个进程结束之后, 将数据添加到 total_data 里面, 最终返回 total_data .