concurrent.futures的使用

2020-12-18 07:27:03 编辑 2 阅读量:363

concurrent.futures 是python3新增加的一个库，用于并发处理，类似于其他语言里的线程池（也有一个进程池），他属于上层的封装，对于用户来说，不用在考虑那么多东西了。

这里主要有：
- Executor：还有它的两个子类ThreadPoolExecutor和ProcessPoolExecutor。两个的用法是类似的，只不过是名字和效果不同，这里介绍ThreadPoolExecutor的用法
- Future：由Executor.submit产生多任务

ThreadPoolExecutor主要有两种使用方式：
#### 1. 使用submit提交任务
简单的例程如下：
```python
# coding: utf-8
from concurrent.futures import ThreadPoolExecutor as Pool
from concurrent.futures import as_completed
import requests

URLS = ['http://qq.com', 'http://sina.com', 'http://www.baidu.com', 'http://c']

def task(url, timeout=10):
    return requests.get(url, timeout=timeout)

# 流程就是新建一个pool，然后用pool添加任务，得到返回的结果，最后注销这个pool，当然也可以留着后面继续使用
# pool可以被重复使用，而且只要不运行.result()不会抛出异常
pool = Pool(max_workers=3)
task1 = pool.submit(task, URLS[0])
task2 = pool.submit(task, URLS[0])
# task3的网址是错误的
task3 = pool.submit(task, URLS[3])
# 如果前面3个task没有完成，那么task4会处理pending的状态
task4 = pool.submit(task, URLS[0])
print(task4.result())
# 在这之前，没有实际调用task3.result(), 没有抛错
try:
    task3.result()
except Exception:
    print("task3 fail")
# 就算pool中有task fail，也可以被shutdown
pool.shutdown()  # 用完之后要自己shutdown，用with的话会系统会自己处理
try:
    task5 = pool.submit(task, URLS[0])
except RuntimeError:
    print("can't use pool after it was shutdown")
# import pdb; pdb.set_trace()

# 用with，这样pool可以自动被注销
with Pool(max_workers=3) as executor:
    future_tasks = [executor.submit(task, url) for url in URLS]
    print(type(future_tasks))

for f in future_tasks:
        if f.running():
            print('%s is running' % str(f))

# 很多地方都可以设置timeout参数，也可以捕捉异常
    # as_completed，先执行的线程先返回
    for f in as_completed(future_tasks, timeout=None):
        try:
            ret = f.done()
            if ret:
                try:
                    f_ret = f.result(timeout=None)
                except Exception as e:
                    print("func run error: %s" % e)
                else:
                    print('%s, done, result: %s, %s' % (str(f), f_ret.url, len(f_ret.content)))
        except Exception as e:
            f.cancel()
            print(str(e))
```

> Executor is an abstract class that provides methods to execute calls asynchronously. It should not be used directly, but through its two subclasses: ThreadPoolExecutor and ProcessPoolExecutor.
官方的介绍直白，用submit注册你的函数，以及要传递的相关的参数,有多个参数也可以传递进去。

```
Executor.submit(fn, *args, **kwargs)
# Schedules the callable to be executed as fn*(*args*, **kwargs) and returns a Future representing the execution of the callable.
with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())
```

在这里，查看任务是否完成除了用as_completed, 也可以用wait。
wait方法接会返回一个tuple(元组)，tuple中包含两个set(集合)，一个是completed(已完成的)另外一个是uncompleted(未完成的)。使用wait方法的一个优势就是获得更大的自由度，它接收三个参数FIRST_COMPLETED, FIRST_EXCEPTION和ALL_COMPLETE，默认设置为ALL_COMPLETED。
```
# coding: utf-8
from concurrent.futures import ThreadPoolExecutor as Pool
from concurrent.futures import wait
import requests

URLS = ['http://qq.com', 'http://sina.com', 'http://www.baidu.com', ]

def task(url, timeout=10):
    return requests.get(url, timeout=timeout)

with Pool(max_workers=3) as executor:
    future_tasks = [executor.submit(task, url) for url in URLS]

for f in future_tasks:
        if f.running():
            print('%s is running' % str(f))

results = wait(future_tasks)
    done = results[0]
    for x in done:
        print(x)
```
wait有timeout和return_when两个参数可以设置。 
- timeout控制wait()方法返回前等待的时间。 
- return_when决定方法什么时间点返回：如果采用默认的ALL_COMPLETED，程序会阻塞直到线程池里面的所有任务都完成；如果采用FIRST_COMPLETED参数，程序并不会等到线程池里面所有的任务都完成。

#### 2. 使用map提交全部的任务
```
# coding: utf-8

from concurrent.futures import ThreadPoolExecutor as Pool
import requests

URLS = ['http://q45354q.com', 'http://www.baidu.com', 'http://sina.com']

def task(url, timeout=10):
    try:
        result = requests.get(url, timeout=timeout)
    except Exception as e:
        result = {}
        print("func error: %s" % e)
    return result

pool = Pool(max_workers=3)
results = pool.map(task, URLS)

# 这里会按照顺序输出, 如果用as_completed则不会按照参数顺序，而是按照完成顺序
# 但是用来map之后，就不能使用as_completed的方法
# 用map要自己在函数里面处理好异常,否则会全部抛出,影响其他线程
for ret in results:
    if ret:
        print('%s, %s' % (ret.url, len(ret.content)))
```

#### 3. 要避免死锁

在用future进行开发时，要避免一定的程度的互相引用。话说，这个有些像gevent的event result，在一个函数中等待另一个结果的完毕。
```

import time
def wait_on_b():
    time.sleep(5)
    print(b.result())  #b不会完成，他一直在等待a的return结果
    return 5
 
def wait_on_a():
    time.sleep(5)
    print(a.result())  #同理a也不会完成，他也是在等待b的结果
    return 6
 
executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)
```
如果上面的任务，一定要同时进行的话，可以开出两个任务，中间用queue通信。注意，要全局变量也不一定安全。

下面的任务也照样会造成死锁的状态，最好不要在已经被submit的函数里面在调用submit。因为这好比你在threading里面再threading一定，是无效的。
```
def wait_on_future():
    f = executor.submit(pow, 5, 2)
    print(f.result())

executor = ThreadPoolExecutor(max_workers=1)
executor.submit(wait_on_future)

def wait_on_future():
    f = executor.submit(pow, 5, 2)
    print(f.result())
 
executor = ThreadPoolExecutor(max_workers=1)
executor.submit(wait_on_future)
```

#### 4. 一些方法的说明
```
class concurrent.futures.Executor
An abstract class that provides methods to execute calls asynchronously. It should not be used directly, but through its concrete subclasses.

submit(fn, *args, **kwargs)
Schedules the callable, fn, to be executed as fn(*args **kwargs) and returns a Future object representing the execution of the callable.

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())
map(func, *iterables, timeout=None, chunksize=1)
Equivalent to map(func, *iterables) except func is executed asynchronously and several calls to func may be made concurrently. The returned iterator raises a concurrent.futures.TimeoutError if __next__() is called and the result isn’t available after timeout seconds from the original call to Executor.map(). timeout can be an int or a float. If timeout is not specified or None, there is no limit to the wait time. If a call raises an exception, then that exception will be raised when its value is retrieved from the iterator. When using ProcessPoolExecutor, this method chops iterables into a number of chunks which it submits to the pool as separate tasks. The (approximate) size of these chunks can be specified by setting chunksize to a positive integer. For very long iterables, using a large value for chunksize can significantly improve performance compared to the default size of 1. With ThreadPoolExecutor, chunksize has no effect.

Changed in version 3.5: Added the chunksize argument.

shutdown(wait=True)
Signal the executor that it should free any resources that it is using when the currently pending futures are done executing. Calls to Executor.submit() and Executor.map() made after shutdown will raise RuntimeError.

If wait is True then this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed. If wait is False then this method will return immediately and the resources associated with the executor will be freed when all pending futures are done executing. Regardless of the value of wait, the entire Python program will not exit until all pending futures are done executing.

You can avoid having to call this method explicitly if you use the with statement, which will shutdown the Executor (waiting as if Executor.shutdown() were called with wait set to True):

Future Objects
The Future class encapulates the asynchronous execution of a callable. Future instances are created by Executor.submit().
用cancel(),可以终止某个线程和进程的任务，返回状态为 True False
Future.cancel()
Attempt to cancel the call. If the call is currently being executed then it cannot be cancelled and the method will return False, otherwise the call will be cancelled and the method will return True.
 
判断是否真的结束了任务。
Future.cancelled()
Return True if the call was successfully cancelled.
 
判断是否还在运行
Future.running()
Return True if the call is currently being executed and cannot be cancelled.
 
判断是正常执行完毕的。 
Future.done()
Return True if the call was successfully cancelled or finished running.
 
针对result结果做超时的控制。 
Future.result(timeout=None)
Return the value returned by the call. If the call hasn’t yet completed then this method will wait up to timeout seconds. If the call hasn’t completed in timeout seconds then a TimeoutError will be raised. timeout can be an int or float.If timeout is not specified or None then there is no limit to the wait time.
# 如果函数是个死循环，中间没有异常，那么就算result出现timeout，那个线程也会持续运行，这里的RuntimeError仅仅指获取result的超时。所以，线程的退出机制要提前设置好

add_done_callback(fn)
Attaches the callable fn to the future. fn will be called, with the future as its only argument, when the future is cancelled or finishes running.
Added callables are called in the order that they were added and are always called in a thread belonging to the process that added them. If the callable raises an Exception subclass, it will be logged and ignored. If the callable raises a BaseException subclass, the behavior is undefined.
If the future has already completed or been cancelled, fn will be called immediately.
```

---
参考：
- [1][concurrent.futures — Launching parallel tasks¶](https://docs.python.org/3/library/concurrent.futures.html)
- [2][使用python的并发库concurrent.futures实现异步](http://xiaorui.cc/2014/11/15/%E4%BD%BF%E7%94%A8python%E7%9A%84%E4%B8%8A%E5%B1%82%E5%B0%81%E8%A3%85%E5%B9%B6%E5%8F%91%E5%BA%93concurrent-futures%E5%AE%9E%E7%8E%B0%E5%BC%82%E6%AD%A5/)
- [3][ python并发之concurrent.futures](http://blog.csdn.net/dutsoft/article/details/54728706)

concurrent.futures的使用

附件: