• 为了保证你在浏览本网站时有着更好的体验,建议使用类似Chrome、Firefox之类的浏览器~~
    • 如果你喜欢本站的内容何不Ctrl+D收藏一下呢,与大家一起分享各种编程知识~
    • 本网站研究机器学习、计算机视觉、模式识别~当然不局限于此,生命在于折腾,何不年轻时多折腾一下

Python grequests闲话

Python admin 来源:SkyRover 3周前 (04-05) 22次浏览 0个评论 扫描二维码

前段时间看到这个grequests库,感觉还是蛮有意思的,所以今天来对这个库拆解拆解。这个库是崇拜的大神kennethreitz写的。Github地址:https://github.com/kennethreitz/grequests

首先看到文档上给的示例:

import grequests


urls = [
'http://www.heroku.com',
'http://python-tablib.org',
'http://httpbin.org',
'http://python-requests.org',
'http://fakedomain/',
'http://kennethreitz.com'
]
# 创建没有发送的request集合
rs = (grequests.get(u) for u in urls)
# 发送
grequests.map(rs)


# 为了防止超时和异常发生,可以指定一个异常处理器
def exception_handler(request, exception):
print("Request failed")


reqs = [
grequests.get('http://httpbin.org/delay/1', timeout=0.001),
grequests.get('http://fakedomain/'),
grequests.get('http://httpbin.org/status/500')]


grequests.map(reqs, exception_handler=exception_handler)


另外,可以使用imap来提高性能

根据这个示例,我们来看看源代码

AsyncRequest

首先来看grequests.get

get = partial(AsyncRequest, 'GET')
options = partial(AsyncRequest, 'OPTIONS')

 

它和其他诸多HTTP方法一样,只是一个快捷方式,其本质是调用了AsyncRequest,看到这个名字就应该知道是异步的Request,所以应该是对普通的Request做了封装和修改:

class AsyncRequest(object):
""" 异步的Request,接收和Session.request相同的参数,还有一些额外的参数
session: 发送请求的session
callback: 在返回对象上的回调函数,和传递hooks={'response': callback}一样
"""
    def __init__(self, method, url, **kwargs):
    #: Request method
      self.method = method
    #: URL to request
      self.url = url
    #: Associated ``Session``
      self.session = kwargs.pop('session', None)
      if self.session is None:
      # requests里的Session对象
        self.session = Session()


      callback = kwargs.pop('callback', None)
      if callback:
        kwargs['hooks'] = {'response': callback}

     #:  The rest arguments for ``Session.request``
      self.kwargs = kwargs
     #: Resulting ``  Response``
      self.response = None

 

可以看到使用partial(AsyncRequest, 'GET')会使得method默认为GET,然后再rs = (grequests.get(u) for u in urls)会形成一个生成器,里面生成AsyncRequest对象。下面来看看这些request是如何发送的:

map

之后程序会调用grequests.map(rs),那么我们来看看map:

def map(requests, stream=False, size=None, exception_handler=None, gtimeout=None):
    """并发的将Requests列表转换成响应
    requests: Request对象的集合
    stream: 如果为True,那么响应内容不会立即下载
    size: 指定同时发起的请求数目,如果为None,就不会有限制
    exception_handler: 回调函数,当异常发生的时候调用,参数是Request和Exception
    gtimeout: Gevent合并所有的超时时间,单位为秒(与每个request的超时时间无关)
    """
    # 将生成器直接转换成list
    requests = list(requests)
    # gevent的Pool对象,是时候研究一波gevent了
    pool = Pool(size) if size else None
    # 调用send函数来发送请求
    jobs = [send(r, pool, stream=stream) for r in requests]
    # 等待所有的greenlet处理单元结束运行
    gevent.joinall(jobs, timeout=gtimeout)
    ret = []
    # 处理所有的请求响应,并且处理异常
    for request in requests:
        if request.response is not None:
            ret.append(request.response)
        # 如果有异常处理器并且request有异常进行处理
        elif exception_handler and hasattr(request, 'exception'):
            ret.append(exception_handler(request, request.exception))
        else:
            # 否则结果置为None
            ret.append(None)
    return ret

 

可以看到map函数很简单,大体流程就是建立了greenlet处理器池,然后对每个request进行调用,然后等待结束,最后得到响应并且处理响应。

# 如果给定size则创建greenlet池,否则为None
pool = Pool(size) if size else None
jobs = [send(r, pool, stream=stream) for r in requests]
gevent.joinall(jobs, timeout=gtimeout)

 

所以问题的关键还是在于gevent那几行的调用,创建管理greenlet的池,用来限制并发,创建好了之后,调用这个pool然后去发送请求,最后等待所有的greenlet结束。

Gevent的并发

无论创建还是没有创建池,最终是要调用send方法的,来看看这个函数:

def send(r, pool=None, stream=False):
    """使用指定的pool发送request对象,如果pool没有指定,这个方法就会阻塞,Pools很有用,因为你可以指定并发限制"""
    if pool is not None:
        return pool.spawn(r.send, stream=stream)
    return gevent.spawn(r.send, stream=stream)

 

gevent.spawn

gevent.spawn创建一个新的Greenlet对象,并且排定运行调用function(*args, **kwargs),这个可以使用gevent.spawn或者是Greenlet.spawn,其实gevent.spawn就是Greenlet.spawn,并且最后会调用Greenlet的类方法,首先实例化一个对象,然后调用start方法,所以也相当于调用Greenlet(*args, **kwargs)

这也是类方法的一个用法,另外的实例化对象的方法

@classmethod
def spawn(cls, *args, **kwargs):
    g = cls(*args, **kwargs)
    g.start()
    return g

pool.spawn

这个方法使用给定的参数开始一个新的greenlet,通常是传递给Greenlet构造函数,并且将其加入这个pool管理的greenlets集合

Pool是Group的子类,提供了限制并发的方法,其spawn方法在greenlets数目达到上限的时候阻塞,直到有一个可用的greenlet。

这个方法也是使用pool实例为Greenlet创建一个实例,然后start it

def spawn(self, *args, **kwargs):
    greenlet = self.greenlet_class(*args, **kwargs)
    self.start(greenlet)
    return greenlet

 

r.send

这个是AsyncRequest的send方法,比较简单,就是发送请求,等待响应。

def send(self, **kwargs):
    merged_kwargs = {}
    merged_kwargs.update(self.kwargs)
    merged_kwargs.update(kwargs)
    try:
        self.response = self.session.request(self.method, self.url, **merged_kwargs)
    except Exception as e:
        self.exception = e
        self.traceback = traceback.format_exc()
    return self

 

imap

imap据说可以提高性能,快来看看吧:

def imap(requests, stream=False, size=2, exception_handler=None):
    """并发的将Request对象的生成器转换成响应的生成器。
    requests: Request对象的生成器
    stream: 如果为True,则不会立即自动下载
    size: 同时发起的请求数,默认为2
    exception_handler: 当发生异常时候回调
    """
    pool = Pool(size)
    def send(r):
        return r.send(stream=stream)
    for request in pool.imap_unordered(send, requests):
        if request.response is not None:
            yield request.response
        elif exception_handler:
            exception_handler(request, request.exception)
    pool.join()

 

可以看到这个函数主要是使用了pool.imap_unordered,其实pool还有一个方法是imap

pool.imap

itertools.imap()是一致的,itertools.imap()可以用于迭代无穷序列,比如itertools.imap(lambda x, y: x * y, [10, 20, 30], itertools.count(1)),如果两个序列长短不一致,以短的为准,并且imap实现了惰性计算,类似生成器。

pool.imap可以并行运行,按顺序从迭代对象中取出元素迭代,应用在函数上,然后收集结果。

如果限制了可以同时进行的greenlets数量,那么最多只有这么多个任务同时进行。

pool.imap_unordered

和imap一样,返回的结果顺序是随意的,比起imap更加轻量级,如果顺序不重要的话,首先应该选用这个。

join

等待这个group的greenlets都运行完,如果这个group没有greenlet的话,立即返回

可以看到,如果不要求顺序的话,imap_unordered会比imap更加高效,同时imap版本肯定比map版本性能好,因为map版本必须全部运行完才能拿数据,但是imap版本只要有greenlet有结果就可以取出来。

小结

这个库就这么多内容,其实主要是使用gevent封装了一层requests,所以核心就是使用gevent,gevent怎么用,如何用,待我继续研究。

不过通过看这个库,也了解到了简单的gevent的用法

 

 


Deeplearn, 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权 , 转载请注明Python grequests闲话
喜欢 (0)
[@lc@]
分享 (0)
admin
关于作者:
互联网行业码农一枚/业余铲屎官/数码影音爱好者/二次元

您必须 登录 才能发表评论!