Python3:concurrent.futures实现高并发。


 

From:https://www.cnblogs.com/weihengblog/p/9812110.html

concurrent.futures 官方文档:https://docs.python.org/3/library/concurrent.futures.html

concurrent.futures: 线程池, 让你更加高效, 并发的处理任务https://www.h3399.cn/201906/703751.html

 

 

python 因为其全局解释器锁 GIL 而无法通过线程实现真正的平行计算。这个论断我们不展开,但是有个概念我们要说明,

IO 密集型 vs 计算密集型:

  1.  IO密集型:读取文件,读取网络套接字频繁。
  2.  计算密集型:大量消耗CPU的数学与逻辑运算,也就是我们这里说的平行计算。

而 concurrent.futures 模块,可以利用 multiprocessing 实现真正的平行计算。

核心原理是:concurrent.futures 会以子进程的形式,平行的运行多个 python 解释器,从而令 python 程序可以利用多核 CPU 来提升执行速度。由于 子进程 与 主解释器 相分离,所以他们的全局解释器锁也是相互独立的。每个子进程都能够完整的使用一个CPU 内核。

 

 

Python 模块 - Concurrent.futures

 

从 Python3.2开始,Python 标准库提供了 concurrent.futures 模块,为开发人员提供了启动异步任务的高级接口。 它提供了 ThreadPoolExecutor 和 ProcessPoolExecutor 两个类,实现了对 threading 和 multiprocessing 的更高级的抽象,对编写 线程池/进程池 提供了直接的支持。 可以将相应的 tasks 直接放入线程池/进程池,不需要维护Queue来操心死锁的问题线程池/进程池会自动帮我们调度。

Future总结

1. python3自带,python2需要安装
 
2. Executer对象
 
它是一个抽象类,它提供了异步执行的方法,他不能直接使用,但可以通过它的子类
 
ThreadPoolExecuter和ProcessPoolExecuter
 
2.1 Executer.submit(fn, *args, **kwargs)
 
fn: 需要异步执行的函数
 
*args,**kwargs fn 接受的参数
 
该方法的作用就是提交一个可执行的回调 task,它返回一个 Future 对象
 
2.2 map(fn, *iterables, timeout=None, chunksize=1)
 
map(task,URLS) # 返回一个 map()迭代器,这个迭代器中的回调执行返回的结果是有序的
 
3. Future对象相关
 
future可以理解为一个在未来完成的操作,这是异步编程的基础
 
通常情况下我们在遇到IO操作的时候,将会发生阻塞,cpu不能做其他事情
 
而future的引入帮助我们在这段等待时间可以完成其他的操作
 
3.1 done():
 
如果当前线程已取消/已成功,返回True。
 
3.2 cance():
 
如果当前线程正在执行,并且不能取消调用,返回Flase。否则调用取消,返回True
 
3.3 running():
 
如果当前的线程正在执行,则返回True
 
3.4 result():
 
返回调用返回的值,如果调用尚未完成,则此方法等待
 
如果等待超时,会抛出concurrent.futures.TimeoutError
 
如果没有指定超时时间,则等待无时间限制
 
如果在完成之前,取消了Future,则会引发CancelledError
 
4. as_completed():
 
在多个Future实例上的迭代器将会被返回
 
这些Future实例由fs完成时产生。
 
由fs返回的任何重复的Future,都会被返回一次。
 
里面保存的都是已经执行完成的Future对象
 
5. wait():
 
返回一个元祖,元祖包含两个元素
 
1. 已完成的future集合
 
2. 未完成的future集合

初体验:

  1. # coding=utf-8
     
    from concurrent import futures
     
    from concurrent.futures import Future
     
    import time
     
     
     
    def return_future(msg):
     
    time.sleep(3)
     
    return msg
     
     
     
     
     
    pool = futures.ThreadPoolExecutor(max_workers=2)
     
     
     
    t1 = pool.submit(return_future,'hello')
     
    t2 = pool.submit(return_future,'world')
     
     
     
    time.sleep(3)
     
    print(t1.done()) # 如果顺利完成,则返回True
     
    time.sleep(3)
     
    print(t2.done())
     
     
     
    print(t1.result()) # 获取future的返回值
     
    time.sleep(3)
     
    print(t2.result())
     
     
     
    print("主线程")

     

     
     

map(func,* iterables,timeout = None,chunksize = 1 )

# coding=utf-8
 
 
 
import time
 
from concurrent.futures import Future,as_completed
 
from concurrent.futures import ThreadPoolExecutor as Pool
 
import requests
 
import time
 
 
 
URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com']
 
 
 
def task(url,timeout=10):
 
return requests.get(url=url,timeout=timeout)
 
 
 
 
 
pool = Pool()
 
result = pool.map(task,URLS)
 
 
 
start_time = time.time()
 
 
 
# 按照 URLS 的顺序返回
 
for res in result:
 
print("{} {}".format(res.url,len(res.content)))
 
 
 
# 无序的
 
with Pool(max_workers=3) as executer:
 
future_task = [executer.submit(task,url) for url in URLS]
 
 
 
for f in as_completed(future_task):
 
if f.done():
 
f_ret = f.result() # f.result()得到task的返回值,requests对象
 
print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content)))
 
 
 
print("耗时", time.time() - start_time)
 
print("主线程")

 

 

Future对象

Future可以理解为一个未来完成的操作
当我们执行io操作的时候,在等待返回结果之前会产生阻塞
cpu不能做其他事情,而Future的引入帮助我们在等待的这段时间可以完成其他操作

from concurrent.futures import ThreadPoolExecutor as Pool
 
from concurrent.futures import as_completed
 
import requests
 
import time
 
 
 
URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com']
 
 
 
def task(url,timeout=10):
 
return requests.get(url=url,timeout=timeout)
 
 
 
# start_time = time.time()
 
# for url in URLS:
 
# ret = task(url)
 
# print("{} {}".format(ret.url,len(ret.content)))
 
# print("耗时",time.time() - start_time)
 
with Pool(max_workers=3) as executor:
 
# 创建future任务
 
future_task = [executor.submit(task,url) for url in URLS]
 
 
 
for f in future_task:
 
if f.running():
 
print("%s is running"%str(f))
 
 
 
for f in as_completed(future_task):
 
try:
 
ret = f.done()
 
if ret:
 
f_ret = f.result()
 
print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content)))
 
except Exception as e:
 
f.cance()
 
print(e)
 
 
 
"""
 
url不是按照顺序返回的,说明并发时,当访问某一个url时,如果没有得到返回结果,不会发生阻塞
 
<Future at 0x1c63990e6d8 state=running> is running
 
<Future at 0x1c639922780 state=running> is running
 
<Future at 0x1c639922d30 state=running> is running
 
<Future at 0x1c63990e6d8 state=finished returned Response>, done, result: http://www.baidu.com/, 2381
 
<Future at 0x1c639922780 state=finished returned Response>, done, result: https://www.qq.com?fromdefault, 243101
 
<Future at 0x1c639922d30 state=finished returned Response>, done, result: http://sina.com/, 23103
 
"""

 

 

模块方法

 

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

wait() 会返回一个tuple,tuple 会包含两个集合:已完成的集合 和 未完成的集合。使用 wait() 会获得更大的自由度,他接受三个参数:FIRST_COMPLETEDFIRST_EXCEPTIONALL_COMPLETE默认为 ALL_COMPLETE

如果采用默认的 ALL_COMPLETED,程序会阻塞直到线程池里面的所有任务都完成,再执行主线程:

 

from concurrent.futures import Future
 
from concurrent.futures import ThreadPoolExecutor as Pool
 
from concurrent.futures import as_completed, wait
 
import requests
 
 
 
URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com']
 
 
 
 
 
def task(url, timeout=10):
 
r = requests.get(url=url, timeout=timeout)
 
print(r.status_code)
 
 
 
 
 
with Pool(max_workers=3) as execute:
 
future_task = [execute.submit(task, url) for url in URLS]
 
 
 
for f in future_task:
 
if f.running():
 
print("%s" % (str(f)))
 
 
 
"""
 
并且wait还有timeout和return_when两个参数
 
return_when有三个常量 (默认是 ALL_COMPLETED)
 
FIRST_COMPLETED 任何一个future_task执行完成时/取消时,改函数返回
 
FIRST_EXCEPTION 任何一个future_task发生异常时,该函数返回,如果没有异常发生,等同于ALL_COMPLETED
 
ALL_COMPLETED 当所有的future_task执行完毕返回。
 
"""
 
results = wait(future_task, return_when="FIRST_COMPLETED") #
 
done = results[0]
 
for d in done:
 
print(d)

 

 

 

 

concurrent.futures.as_completed(fs, timeout=None)

在多个 Future 实例上的迭代器将会被返回,这些 Future 实例由 fs 完成时产生。由 fs 返回的任何重复的 Future,都会被返回一次。里面保存的都是已经执行完成的 Future 对象。

from concurrent.futures import ThreadPoolExecutor as Pool
 
from concurrent.futures import as_completed
 
import requests
 
import time
 
 
 
URLS = ['http://www.baidu.com', 'http://qq.com', 'http://sina.com']
 
 
 
def task(url,timeout=10):
 
return requests.get(url=url,timeout=timeout)
 
 
 
with Pool(max_workers=3) as executor:
 
# 创建future任务
 
future_task = [executor.submit(task,url) for url in URLS]
 
 
 
for f in future_task:
 
if f.running():
 
print("%s is running"%str(f))
 
 
 
for f in as_completed(future_task):
 
try:
 
ret = f.done()
 
if ret:
 
f_ret = f.result()
 
print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content)))
 
except Exception as e:
 
f.cance()
 
print(e)

 

下面我们将学习 concurrent.futures 模块中的类。concurrent.futures 基础模块是 executor 和 future。

使用示例代码:

# -*- coding:utf-8 -*-
 
 
 
import redis
 
from redis import WatchError
 
from concurrent.futures import ProcessPoolExecutor
 
 
 
r = redis.Redis(host='127.0.0.1', port=6379)
 
 
 
 
 
# 减库存函数, 循环直到减库存完成
 
# 库存充足, 减库存成功, 返回True
 
# 库存不足, 减库存失败, 返回False
 
 
 
def reduce_stock():
 
 
 
# python中redis事务是通过pipeline的封装实现的
 
with r.pipeline() as pipe:
 
while True:
 
try:
 
# watch库存键, multi后如果该key被其他客户端改变, 事务操作会抛出WatchError异常
 
pipe.watch('stock:count')
 
count = int(pipe.get('stock:count'))
 
if count > 0: # 有库存
 
# 事务开始
 
pipe.multi()
 
pipe.decr('stock:count')
 
# 把命令推送过去
 
# execute返回命令执行结果列表, 这里只有一个decr返回当前值
 
print(pipe.execute()[0])
 
return True
 
else:
 
return False
 
except WatchError as ex:
 
# 打印WatchError异常, 观察被watch锁住的情况
 
print(ex)
 
pipe.unwatch()
 
 
 
 
 
def worker():
 
while True:
 
# 没有库存就退出
 
if not reduce_stock():
 
break
 
 
 
 
 
if __name__ == "__main__":
 
# 设置库存为100
 
r.set("stock:count", 100)
 
 
 
# 多进程模拟多个客户端提交
 
with ProcessPoolExecutor() as pool:
 
for _ in range(10):
 
pool.submit(worker)

 

 

 

concurrent.futures 模块详解

 

1. Executor对象

class concurrent.futures.Executor

Executor是一个抽象类,它提供了异步执行调用的方法。它不能直接使用,但可以通过它的两个子类ThreadPoolExecutor或者ProcessPoolExecutor进行调用。

 

1.1 Executor.submit(fn, *args, **kwargs)

fn:需要异步执行的函数
*args, **kwargs:fn 的参数

示例代码:

# -*- coding:utf-8 -*-
 
from concurrent import futures
 
 
 
 
 
def test(num):
 
import time
 
return time.ctime(), num
 
 
 
 
 
with futures.ThreadPoolExecutor(max_workers=1) as executor:
 
future = executor.submit(test, 1)
 
print(future.result())

 

1.2 Executor.map(func, *iterables, timeout=None)

相当于map(func, *iterables),但是func是异步执行。timeout的值可以是int或float,如果操作超时,会返回raisesTimeoutError;如果不指定timeout参数,则不设置超时间。

func:需要异步执行的函数
*iterables:可迭代对象,如列表等。每一次func执行,都会从iterables中取参数。
timeout:设置每次异步操作的超时时间

示例代码:

# -*- coding:utf-8 -*-
 
from concurrent import futures
 
 
 
 
 
def test(num):
 
import time
 
return time.ctime(), num
 
 
 
 
 
data = [1, 2, 3]
 
with futures.ThreadPoolExecutor(max_workers=1) as executor:
 
for future in executor.map(test, data):
 
print(future)

 

1.3 Executor.shutdown(wait=True)

释放系统资源,在Executor.submit()或 Executor.map()等异步操作后调用。使用with语句可以避免显式调用此方法

 

2. ThreadPoolExecutor对象

ThreadPoolExecutor类是Executor子类,使用线程池执行异步调用.

class concurrent.futures.ThreadPoolExecutor(max_workers),使用 max_workers 数目的线程池执行异步调用

 

3. ProcessPoolExecutor对象

ThreadPoolExecutor类是Executor子类,使用进程池执行异步调用.

class concurrent.futures.ProcessPoolExecutor(max_workers=None),使用 max_workers数目的进程池执行异步调用,如果max_workers为None则使用机器的处理器数目(如4核机器max_worker配置为None时,则使用4个进程进行异步并发)。

示例代码:

# -*- coding:utf-8 -*-
 
from concurrent import futures
 
 
 
 
 
def test(num):
 
import time
 
return time.ctime(), num
 
 
 
 
 
def muti_exec(m, n):
 
# m 并发次数
 
# n 运行次数
 
 
 
with futures.ProcessPoolExecutor(max_workers=m) as executor: # 多进程
 
# with futures.ThreadPoolExecutor(max_workers=m) as executor: #多线程
 
executor_dict = dict((executor.submit(test, times), times) for times in range(m * n))
 
 
 
for future in futures.as_completed(executor_dict):
 
times = executor_dict[future]
 
if future.exception() is not None:
 
print('%r generated an exception: %s' % (times, future.exception()))
 
else:
 
print('RunTimes:%d,Res:%s' % (times, future.result()))
 
 
 
 
 
if __name__ == '__main__':
 
muti_exec(5, 1)

 

 

调度单个任务

 

执行者类Executor调度单个任务,使用submit() 函数,然后用返回的 Future 实例等待任务结果。

Executor 是一个 Python concurrent.futures 模块的抽象类。 它不能直接使用,我们需要使用以下具体子类之一 -

  • ThreadPoolExecutor:线程池
  • ProcessPoolExecutor:进程池

 

示例代码:

from concurrent import futures
 
import time
 
import random
 
 
 
 
 
def task(n):
 
time.sleep(random.randint(1, 10))
 
return n
 
 
 
 
 
executor = futures.ThreadPoolExecutor(max_workers=3)
 
future = executor.submit(task, 5)
 
print('future: {}'.format(future))
 
result = future.result()
 
print('result: {}'.format(result))

 

 

线程池 和 进程池

 

  1. ThreadPoolExecutor 是 Executor类的具体子类之一。 子类使用多线程,我们得到一个提交任务的线程池。 该池将任务分配给可用线程并安排它们运行。
  2. ProcessPoolExecutor Executor类的具体子类之一。 它使用多重处理,并且我们获得提交任务的进程池。 此池将任务分配给可用的进程并安排它们运行。

 

如何创建一个 ThreadPoolExecutor 或者 ProcessPoolExecutor?

        在concurrent.futures模块及其具体子类Executor的帮助下,可以很容易地创建一个线程池或者进程池。 需要使用我们想要的池中的线程数构造一个ThreadPoolExecutor 或者 ProcessPoolExecutor。 默认情况下,数字是5。然后可以提交一个任务到线程池或者进程池。 当submit()任务时,会返回Future对象。 Future对象有一个名为done()的方法,它告诉Future是否已经解决。 有了这个,为这个特定的Future对象设定了一个值。 当任务完成时,线程池执行器将该值设置为Future的对象。

线程池 示例代码:

from concurrent.futures import ThreadPoolExecutor
 
from time import sleep
 
 
 
 
 
def task(message):
 
sleep(2)
 
return message
 
 
 
 
 
def main():
 
executor = ThreadPoolExecutor(5)
 
future = executor.submit(task, "Completed")
 
print(future.done())
 
sleep(2)
 
print(future.done())
 
print(future.result())
 
 
 
 
 
if __name__ == '__main__':
 
main()

 

结果截图:

在上面的例子中,一个ThreadPoolExecutor已经由5个线程构造而成。 然后,在提供消息之前等待2秒的任务被提交给线程池执行器。 从输出中可以看出,任务直到2秒才完成,所以第一次调用done()将返回False2秒后,任务完成,我们通过调用result()方法得到future的结果。

进程池 示例代码:

from concurrent.futures import ProcessPoolExecutor
 
from time import sleep
 
 
 
 
 
def task(message):
 
sleep(2)
 
return message
 
 
 
 
 
def main():
 
executor = ProcessPoolExecutor(5)
 
future = executor.submit(task, ("Completed"))
 
print(future.done())
 
sleep(2)
 
print(future.done())
 
print(future.result())
 
 
 
 
 
if __name__ == '__main__':
 
main()

 

 

实例化ThreadPoolExecutor 或者 ProcessPoolExecutor  之 上下文管理器
另一种实例化ThreadPoolExecutor的方法是在上下文管理器的帮助下完成的。 它的工作方式与上例中使用的方法类似。 使用上下文管理器的主要优点是它在语法上看起来不错。 实例化可以在下面的代码的帮助下完成

with ThreadPoolExecutor(max_workers = 5) as executor
 
或者
 
with ProcessPoolExecutor(max_workers = 5) as executor

 

示例

以下示例是从 Python 文档借用的。 在这个例子中,首先必须导入 concurrent.futures 模块。 然后创建一个名为 load_url()的函数,它将加载请求的url。 然后该函数用池中的5个线程创建 ThreadPoolExecutorThreadPoolExecutor 已被用作上下文管理器。 我们可以通过调用 result()方法来获得 future的结果。

import concurrent.futures
 
import urllib.request
 
 
 
URLS = [
 
'http://www.foxnews.com/',
 
'https://www.yiibai.com/',
 
'http://europe.wsj.com/',
 
'http://www.bbc.co.uk/',
 
'http://some-made-up-domain.com/'
 
]
 
 
 
 
 
def load_url(url, timeout):
 
with urllib.request.urlopen(url, timeout=timeout) as conn:
 
return conn.read()
 
 
 
 
 
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
 
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
 
for future in concurrent.futures.as_completed(future_to_url):
 
url = future_to_url[future]
 
try:
 
data = future.result()
 
except Exception as exc:
 
print('%r generated an exception: %s' % (url, exc))
 
else:
 
print('%r page is %d bytes' % (url, len(data)))

 

以下将是上面的Python脚本的输出 -

  1.  
    'http: //some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
  2.  
    'http: //www.foxnews.com/' page is 229313 bytes
  3.  
    'http: //www.yiibai.com/' page is 168933 bytes
  4.  
    'http: //www.bbc.co.uk/' page is 283893 bytes
  5.  
    'http: //europe.wsj.com/' page is 938109 bytes

进程池:

import concurrent.futures
 
from concurrent.futures import ProcessPoolExecutor
 
import urllib.request
 
 
 
URLS = ['http://www.foxnews.com/',
 
'http://www.cnn.com/',
 
'http://europe.wsj.com/',
 
'http://www.bbc.co.uk/',
 
'http://some-made-up-domain.com/']
 
 
 
 
 
def load_url(url, timeout):
 
with urllib.request.urlopen(url, timeout=timeout) as conn:
 
return conn.read()
 
 
 
 
 
def main():
 
with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
 
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
 
for future in concurrent.futures.as_completed(future_to_url):
 
url = future_to_url[future]
 
try:
 
data = future.result()
 
except Exception as exc:
 
print('%r generated an exception: %s' % (url, exc))
 
else:
 
print('%r page is %d bytes' % (url, len(data)))
 
 
 
 
 
if __name__ == '__main__':
 
main()

 

 

 

使用 map() 调度多任务,有序返回

 

使用map(),多个worker并发地从输入迭代器里取数据,处理,然后按顺序返回结果。

示例代码:

from concurrent import futures
 
import time
 
import random
 
 
 
 
 
def task(n):
 
time.sleep(random.randint(1, 10))
 
return n
 
 
 
 
 
executor = futures.ThreadPoolExecutor(max_workers=3)
 
results = executor.map(task, range(1, 10))
 
print('unprocessed results: {}'.format(results))
 
real_results = list(results)
 
print('real results: {}'.format(real_results))

 

使用 Executor.map() 函数

Python map()函数广泛用于许多任务。 一个这样的任务是对可迭代内的每个元素应用某个函数。 同样,可以将迭代器的所有元素映射到一个函数,并将这些作为独立作业提交到ThreadPoolExecutor之外。 考虑下面的Python脚本示例来理解函数的工作原理。

示例
在下面的示例中,map函数用于将square()函数应用于values数组中的每个值。

from concurrent.futures import ThreadPoolExecutor
 
from concurrent.futures import as_completed
 
 
 
values = [2, 3, 4, 5]
 
 
 
 
 
def square(n):
 
return n * n
 
 
 
 
 
def main():
 
with ThreadPoolExecutor(max_workers=3) as executor:
 
results = executor.map(square, values)
 
for result in results:
 
print(result)
 
 
 
 
 
if __name__ == '__main__':
 
main()

 

以下将是上面的Python脚本的输出 :

进程池:

from concurrent.futures import ProcessPoolExecutor
 
from concurrent.futures import as_completed
 
 
 
values = [2, 3, 4, 5]
 
 
 
 
 
def square(n):
 
return n * n
 
 
 
 
 
def main():
 
with ProcessPoolExecutor(max_workers=3) as executor:
 
results = executor.map(square, values)
 
for result in results:
 
print(result)
 
 
 
 
 
if __name__ == '__main__':
 
main()

 

 

多任务调度,无序返回

 

不断将任务submit到executor,返回future列表,使用as_completed无序产生每个任务的结果。

示例代码:

from concurrent import futures
 
import time
 
import random
 
 
 
 
 
def task(n):
 
time.sleep(random.randint(1, 10))
 
return n
 
 
 
 
 
executor = futures.ThreadPoolExecutor(max_workers=3)
 
future_list = [executor.submit(task, i) for i in range(1, 10)]
 
for f in futures.as_completed(future_list):
 
print(f.result())

 

何时使用ProcessPoolExecutor 和 ThreadPoolExecutor ?

现在我们已经学习了两个Executor类 - ThreadPoolExecutorProcessPoolExecutor,我们需要知道何时使用哪个执行器。需要在受CPU限制的工作负载情况下选择ProcessPoolExecutor,而在受I/O限制的工作负载情况下则需要选择ThreadPoolExecutor

如果使用ProcessPoolExecutor,那么不需要担心GIL,因为它使用多处理。 而且,与ThreadPoolExecution相比,执行时间会更少。

 

转载:https://blog.csdn.net/freeking101/article/details/97395745

 

 


免责声明!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系本站邮箱yoyou2525@163.com删除。



 
粤ICP备18138465号  © 2018-2025 CODEPRJ.COM