Python 简单并发的代码记录

简单介绍

近期想优化一下API的请求，顺带测试一下并发编程能快多少，用到进程、线程，顺带保留一些直接可用的简单的并发代码，方便后期复制粘贴

1、urllib3 网络请求
2、获取函数执行时间（大致的）
3、进程的简单实现
4、线程的简单实现
5、通过CPU计算以及网络IO请求，简单的对比性能上的差异

公共代码

from datetime import datetime
import urllib3
import jsonmanager  = urllib3.PoolManager(num_pools=5)def request_api(site):# print(site)test_url = 'http://127.0.0.1:9000/'r = manager.request('GET', test_url)data = r.data# json.loads(data)return datadef cpu_test(number):result = 235 * numberreturn resultdef print_time(func1, func2):start_time = datetime.now()func1(func2)end_time = datetime.now()print(func1, end_time-start_time)

线程测试适合 IO密集型：读写

from concurrent.futures import ThreadPoolExecutordef thread_map_test(func):result = []site_list = range(1000)with ThreadPoolExecutor(max_workers=12) as executor:# 返回结果是一个迭代器ans = executor.map(func, site_list)for res in ans:result.append(res)return resultdef thread_test(func):result = []site_list = range(1000)with ThreadPoolExecutor(max_workers=5) as executor:ans = [executor.submit(func, i) for i in site_list ]# as_completed(ans)for res in ans:result.append(res.result())

进程测试适合CPU密集型：计算

import multiprocessingdef process_by_map(func):data_list = range(1000)with multiprocessing.Pool(processes=4) as pool: result = pool.map(func, data_list)return result # 会阻塞
def process_by_apply(func):result = []data_list = range(1000)with multiprocessing.Pool(processes=4) as pool: for line in data_list:result.append(pool.apply(func, (line,)))return result# 异步非阻塞
def process_by_apply_async(func):result = []data_list = range(1000)with multiprocessing.Pool(processes=4) as pool: for line in data_list:result.append(pool.apply_async(func, (line,)))return result

综合测试

注意：多进程必须要写if name == ‘main’: 否则，IDE上面啥也不提示，但是没执行结果，直接点击执行就会提醒有异常

# 请注意，多进程必须要写if __name__ == '__main__': 
if __name__ == '__main__':print('''计算测试 100万次计算''')# 0:00:00.362031print_time(process_by_map, cpu_test)# 0:01:48.696417print_time(process_by_apply, cpu_test)# 0:00:11.481112print_time(process_by_apply_async, cpu_test)# 0:00:25.352362print_time(thread_map_test, cpu_test)# 0:00:20.073205print_time(thread_test, cpu_test)print('''网络IO 测试  1000次请求''')# 0:00:04.105110print_time(process_by_map, request_api)# 0:00:07.180077print_time(process_by_apply, request_api)# 0:00:00.139999print_time(process_by_apply_async, request_api)# 0:00:02.757811print_time(thread_map_test, request_api)# 0:00:02.712561print_time(thread_test, request_api)