网络爬虫必备知识之concurrent.futures库

1.concurrent.futures库简介python标准库为我们提供了threading和mutiprocessing模块实现异步多线程/多进程功能。从python3.2版本开始,标准库又为

大家好,又见面了,我是全栈君,今天给大家准备了Idea注册码。

就库的范围,个人认为网络爬虫必备库知识包括urllib、requests、re、BeautifulSoup、concurrent.futures,接下来将结对concurrent.futures库的使用方法进行总结

建议阅读本博的博友先阅读下上篇博客:
python究竟要不要使用多线程,将会对concurrent.futures库的使用有帮助。

1. concurrent.futures库简介

  python标准库为我们提供了threading和mutiprocessing模块实现异步多线程/多进程功能。从python3.2版本开始,标准库又为我们提供了concurrent.futures模块来实现线程池和进程池功能,实现了对threading和mutiprocessing模块的高级抽象,更大程度上方便了我们python程序员。

  concurrent.futures模块提供了ThreadPoolExecutorProcessPoolExecutor两个类

(1)看下来个类的继承关系和关键属性

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

print('ThreadPoolExecutor继承关系:',ThreadPoolExecutor.__mro__)
    print('ThreadPoolExecutor属性:',[attr for attr in dir(ThreadPoolExecutor) if not attr.startswith('_')])
    print('ProcessPoolExecutor继承关系:',ProcessPoolExecutor.__mro__)
    print('ThreadPoolExecutor属性:',[attr for attr in dir(ProcessPoolExecutor) if not attr.startswith('_')])

网络爬虫必备知识之concurrent.futures库

  都继承自futures._base.Executor类,拥有三个重要方法map、submit和shutdow,这样看起来就很简单了

(2)再看下futures._base.Executor基类实现

网络爬虫必备知识之concurrent.futures库
网络爬虫必备知识之concurrent.futures库

class Executor(object):
    """This is an abstract base class for concrete asynchronous executors."""

    def submit(self, fn, *args, **kwargs):
        """Submits a callable to be executed with the given arguments.

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

        Returns:
            A Future representing the given call.
        """
        raise NotImplementedError()

    def map(self, fn, *iterables, timeout=None, chunksize=1):
        """Returns an iterator equivalent to map(fn, iter).

        Args:
            fn: A callable that will take as many arguments as there are
                passed iterables.
            timeout: The maximum number of seconds to wait. If None, then there
                is no limit on the wait time.
            chunksize: The size of the chunks the iterable will be broken into
                before being passed to a child process. This argument is only
                used by ProcessPoolExecutor; it is ignored by
                ThreadPoolExecutor.

        Returns:
            An iterator equivalent to: map(func, *iterables) but the calls may
            be evaluated out-of-order.

        Raises:
            TimeoutError: If the entire result iterator could not be generated
                before the given timeout.
            Exception: If fn(*args) raises for any values.
        """
        if timeout is not None:
            end_time = timeout + time.time()

        fs = [self.submit(fn, *args) for args in zip(*iterables)]

        # Yield must be hidden in closure so that the futures are submitted
        # before the first iterator value is required.
        def result_iterator():
            try:
                # reverse to keep finishing order
                fs.reverse()
                while fs:
                    # Careful not to keep a reference to the popped future
                    if timeout is None:
                        yield fs.pop().result()
                    else:
                        yield fs.pop().result(end_time - time.time())
            finally:
                for future in fs:
                    future.cancel()
        return result_iterator()

    def shutdown(self, wait=True):
        """Clean-up the resources associated with the Executor.

        It is safe to call this method several times. Otherwise, no other
        methods can be called after this one.

        Args:
            wait: If True then shutdown will not return until all running
                futures have finished executing and the resources used by the
                executor have been reclaimed.
        """
        pass

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown(wait=True)
        return False

View Code

  提供了map、submit、shutdow和with方法,下面首先对这个几个方法的使用进行说明

2. map函数

  函数原型:def map(self, fn, *iterables, timeout=None, chunksize=1)

  map函数和python自带的map函数用法一样,只不过该map函数从迭代器获取参数后异步执行,timeout用于设置超时时间

  参数chunksize的理解

The size of the chunks the iterable will be broken into
 before being passed to a child process. This argument is only
 used by ProcessPoolExecutor; it is ignored by ThreadPoolExecutor.

  例:

from concurrent.futures import ThreadPoolExecutor
import time 
import requests

def download(url):
    headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0',
                'Connection':'keep-alive',
                'Host':'example.webscraping.com'}
    response = requests.get(url, headers=headers)
    return(response.status_code)
    
if __name__ == '__main__':
    urllist = ['http://example.webscraping.com/places/default/view/Afghanistan-1',
               'http://example.webscraping.com/places/default/view/Aland-Islands-2']
               
    pool = ProcessPoolExecutor(max_workers = 2) 
start
= time.time() result = list(pool.map(download, urllist)) end = time.time() print('status_code:',result) print('使用多线程--timestamp:{:.3f}'.format(end-start))

3. submit函数

  函数原型:def submit(self, fn, *args, **kwargs)

  fn:需要异步执行的函数

  args、kwargs:函数传递的参数

  :下例中future类的使用的as_complete后面介绍

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time 
import requests

def download(url):
    headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0',
                'Connection':'keep-alive',
                'Host':'example.webscraping.com'}
    response = requests.get(url, headers=headers)
    return response.status_code
    
if __name__ == '__main__':
    urllist = ['http://example.webscraping.com/places/default/view/Afghanistan-1',
               'http://example.webscraping.com/places/default/view/Aland-Islands-2']
               
    start = time.time()               
    pool = ProcessPoolExecutor(max_workers = 2)                  
    futures = [pool.submit(download,url) for url in urllist]
    for future in futures:
        print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
    print('#### 分界线 ####')
    for future in as_completed(futures, timeout=2):
        print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
        print(future.result())
    end = time.time()
    print('使用多线程--timestamp:{:.3f}'.format(end-start))

   输出

  网络爬虫必备知识之concurrent.futures库

4. shutdown函数

  函数原型:def shutdown(self, wait=True)

  此函数用于释放异步执行操作后的系统资源

  由于_base.Executor类提供了上下文方法,将shutdown封装在了__exit__中,若使用with方法,将不需要自己进行资源释放

with ProcessPoolExecutor(max_workers = 2) as pool:

5. Future类

  submit函数返回Future对象,Future类提供了跟踪任务执行状态的方法:

  future.running():判断任务是否执行

  futurn.done:判断任务是否执行完成

  futurn.result():返回函数执行结果

futures = [pool.submit(download,url) for url in urllist]
for future in futures:
    print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
print('#### 分界线 ####')
for future in as_completed(futures, timeout=2):
    print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
    print(future.result())

  as_completed方法传入futures迭代器和timeout两个参数

  默认timeout=None,阻塞等待任务执行完成,并返回执行完成的future对象迭代器,迭代器是通过yield实现的。 

  timeout>0,等待timeout时间,如果timeout时间到仍有任务未能完成,不再执行并抛出异常TimeoutError

6. 回调函数

  Future类提供了add_done_callback函数可以自定义回调函数:

def add_done_callback(self, fn):
        """Attaches a callable that will be called when the future finishes.

        Args:
            fn: A callable that will be called with this future as its only
                argument when the future completes or is cancelled. The callable
                will always be called by a thread in the same process in which
                it was added. If the future has already completed or been
                cancelled then the callable will be called immediately. These
                callables are called in the order that they were added.
        """
        with self._condition:
            if self._state not in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
                self._done_callbacks.append(fn)
                return
        fn(self)

   例子:

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time 
import requests

def download(url):
    headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0',
                'Connection':'keep-alive',
                'Host':'example.webscraping.com'}
    response = requests.get(url, headers=headers)
    return response.status_code
    
def callback(future): print(future.result()) if __name__ == '__main__':
    urllist = ['http://example.webscraping.com/places/default/view/Afghanistan-1',
               'http://example.webscraping.com/places/default/view/Aland-Islands-2',
               'http://example.webscraping.com/places/default/view/Albania-3',
               'http://example.webscraping.com/places/default/view/Algeria-4',
               'http://example.webscraping.com/places/default/view/American-Samoa-5']
               
    start = time.time()               
    with ProcessPoolExecutor(max_workers = 2) as pool:                  
        futures = [pool.submit(download,url) for url in urllist]
        for future in futures:
            print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
            print('#### 分界线 ####')
        for future in as_completed(futures, timeout=5):
            future.add_done_callback(callback)
            print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
        end = time.time()
        print('使用多线程--timestamp:{:.3f}'.format(end-start))

7. wait函数

  函数原型:def wait(fs, timeout=None, return_when=ALL_COMPLETED)

网络爬虫必备知识之concurrent.futures库
网络爬虫必备知识之concurrent.futures库

def wait(fs, timeout=None, return_when=ALL_COMPLETED):
    """Wait for the futures in the given sequence to complete.

    Args:
        fs: The sequence of Futures (possibly created by different Executors) to
            wait upon.
        timeout: The maximum number of seconds to wait. If None, then there
            is no limit on the wait time.
        return_when: Indicates when this function should return. The options
            are:

            FIRST_COMPLETED - Return when any future finishes or is
                              cancelled.
            FIRST_EXCEPTION - Return when any future finishes by raising an
                              exception. If no future raises an exception
                              then it is equivalent to ALL_COMPLETED.
            ALL_COMPLETED -   Return when all futures finish or are cancelled.

    Returns:
        A named 2-tuple of sets. The first set, named 'done', contains the
        futures that completed (is finished or cancelled) before the wait
        completed. The second set, named 'not_done', contains uncompleted
        futures.
    """
    with _AcquireFutures(fs):
        done = set(f for f in fs
                   if f._state in [CANCELLED_AND_NOTIFIED, FINISHED])
        not_done = set(fs) - done

        if (return_when == FIRST_COMPLETED) and done:
            return DoneAndNotDoneFutures(done, not_done)
        elif (return_when == FIRST_EXCEPTION) and done:
            if any(f for f in done
                   if not f.cancelled() and f.exception() is not None):
                return DoneAndNotDoneFutures(done, not_done)

        if len(done) == len(fs):
            return DoneAndNotDoneFutures(done, not_done)

        waiter = _create_and_install_waiters(fs, return_when)

    waiter.event.wait(timeout)
    for f in fs:
        with f._condition:
            f._waiters.remove(waiter)

    done.update(waiter.finished_futures)
    return DoneAndNotDoneFutures(done, set(fs) - done)

View Code

  wait方法返回一个中包含两个元组,元组中包含两个集合(set),一个是已经完成的(completed),一个是未完成的(uncompleted)

  它接受三个参数,重点看下第三个参数:

  FIRST_COMPLETED:Return when any future finishes or iscancelled. 

  FIRST_EXCEPTION:Return when any future finishes by raising an exception,
             If no future raises an exception then it is equivalent to ALL_COMPLETED.
  ALL_COMPLETED:Return when all futures finish or are cancelled
  例:
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,\
            as_completed,wait,ALL_COMPLETED, FIRST_COMPLETED, FIRST_EXCEPTION
import time 
import requests

def download(url):
    headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:63.0) Gecko/20100101 Firefox/63.0',
                'Connection':'keep-alive',
                'Host':'example.webscraping.com'}
    response = requests.get(url, headers=headers)
    return response.status_code
    
if __name__ == '__main__':
    urllist = ['http://example.webscraping.com/places/default/view/Afghanistan-1',
               'http://example.webscraping.com/places/default/view/Aland-Islands-2',
               'http://example.webscraping.com/places/default/view/Albania-3',
               'http://example.webscraping.com/places/default/view/Algeria-4',
               'http://example.webscraping.com/places/default/view/American-Samoa-5']
               
    start = time.time()               
    with ProcessPoolExecutor(max_workers = 2) as pool:                  
        futures = [pool.submit(download,url) for url in urllist]
        for future in futures:
            print('执行中:%s, 已完成:%s' % (future.running(), future.done()))
        print('#### 分界线 ####')
        completed, uncompleted = wait(futures, timeout=2, return_when=FIRST_COMPLETED)
        for cp in completed:
            print('执行中:%s, 已完成:%s' % (cp.running(), cp.done()))
            print(cp.result())
        end = time.time()
        print('使用多线程--timestamp:{:.3f}'.format(end-start))

  输出

 网络爬虫必备知识之concurrent.futures库

  只返回了一个完成的

 

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/120045.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • pycharm升级pip版本_怎么升级pip版本

    pycharm升级pip版本_怎么升级pip版本python-mpipinstall–upgradepip

    2022年10月23日
  • 使用 JSONPath 解析 JSON 完整内容详解

    使用 JSONPath 解析 JSON 完整内容详解jsonpath的介绍:JsonPath是一种简单的方法来提取给定JSON文档的部分内容。JsonPath有许多编程语言,如Javascript,Python和PHP,Java。JsonPath提供的json解析非常强大,它提供了类似正则表达式的语法,基本上可以满足所有你想要获得的json内容。 github上有它的应用:https://github.com/json-path/…

  • 使用java代码打印三角形、平行四边形、菱形

    使用java代码打印三角形、平行四边形、菱形

  • 伪装计算机主机,位置伪装大师电脑版

    伪装计算机主机,位置伪装大师电脑版《位置伪装大师电脑版》是一款免费的GPS位置变换软件,《位置伪装大师电脑版》能够进行GPS位置模拟,让你轻松变换自己的位置,变换位置随心所欲!官方介绍位置伪装大师v3.6新版来袭!变换位置,随心所欲!!全新的界面,全新的功能,全新的体验。更加简洁、更加人性化的操作流程。功能介绍-支持国外伪装,实现全球伪装-一键收藏地点,方便快捷-多种搜索模式,可以快速找到位置-支持经纬度定位,精确寻找位置…

  • 爬虫工具_应用程序market

    爬虫工具_应用程序market一个简单的异步爬虫.私信太多,统一回答一下:关于异步函数的:1.真正派发任务的是consumer这个coroutine,所以也在内部做了并发控制.2.process_content用于获取html及保存到mysql.关于异步相关(asyncio)的:1.await相当于yieldfrom.2.await后面是一个coroutine,…

    2022年10月25日
  • URI和URL的区别比较与理解[通俗易懂]

    URI和URL的区别比较与理解[通俗易懂]一、URI<1>什么是URIURI,通一资源标志符(UniformResourceIdentifier,URI),表示的是web上每一种可用的资源,如HTML文档、图像、视频片段、程序等都由一个URI进行定位的。<2>URI的结构组成URI通常由三部分组成:①访问资源的命名机制;②存放资源的主机名;③资源自身的名称。<3>…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号