限制/限制 GRequests 中 HTTP 请求的速率答案

【问题标题】：Limiting/throttling the rate of HTTP requests in GRequests限制/限制 GRequests 中 HTTP 请求的速率
【发布时间】：2013-12-13 09:11:53
【问题描述】：

我正在用 Python 2.7.3 编写一个带有 GRequests 和 lxml 的小脚本，这将使我能够从各种网站收集一些收藏卡价格并进行比较。问题是其中一个网站限制了请求的数量，如果我超过它，就会发回 HTTP 错误 429。

有没有办法限制 GRequestes 中的请求数，这样我就不会超过我指定的每秒请求数？另外 - 如果发生 HTTP 429，我如何让 GRequestes 在一段时间后重试？

附带说明 - 他们的限制低得离谱。每 15 秒有 8 个请求。我多次使用浏览器破坏了它，只是刷新了等待价格变化的页面。

【问题讨论】：

您是否使用 grequests 在他们的网站上注册了多个 url？如果是这样，最好同步向他们的网站发出请求，这样您就可以轻松控制重试
我同意@dm03514 - 你不需要 grequests。我最近读到的一种算法可以针对这类问题进行优化。
我正在尝试同时从同一个站点访问多个 url，因为我希望能够同时监控多张卡的价格。
@WayneWerner 这听起来很有趣。你介意分享你的同步方法吗？

标签： python http python-requests throttling rate-limiting

【解决方案1】：

要回答我自己的问题，因为我必须自己解决这个问题，而且关于这个问题的信息似乎很少。

思路如下。与 GRequests 一起使用的每个请求对象在创建时都可以将会话对象作为参数。另一方面，会话对象可以安装在发出请求时使用的 HTTP 适配器。通过创建我们自己的适配器，我们可以拦截请求并以我们认为最适合我们的应用程序的方式限制它们。就我而言，我最终得到了下面的代码。

用于节流的对象：

DEFAULT_BURST_WINDOW = datetime.timedelta(seconds=5)
DEFAULT_WAIT_WINDOW = datetime.timedelta(seconds=15)


class BurstThrottle(object):
    max_hits = None
    hits = None
    burst_window = None
    total_window = None
    timestamp = None

    def __init__(self, max_hits, burst_window, wait_window):
        self.max_hits = max_hits
        self.hits = 0
        self.burst_window = burst_window
        self.total_window = burst_window + wait_window
        self.timestamp = datetime.datetime.min

    def throttle(self):
        now = datetime.datetime.utcnow()
        if now < self.timestamp + self.total_window:
            if (now < self.timestamp + self.burst_window) and (self.hits < self.max_hits):
                self.hits += 1
                return datetime.timedelta(0)
            else:
                return self.timestamp + self.total_window - now
        else:
            self.timestamp = now
            self.hits = 1
            return datetime.timedelta(0)

HTTP 适配器：

class MyHttpAdapter(requests.adapters.HTTPAdapter):
    throttle = None

    def __init__(self, pool_connections=requests.adapters.DEFAULT_POOLSIZE,
                 pool_maxsize=requests.adapters.DEFAULT_POOLSIZE, max_retries=requests.adapters.DEFAULT_RETRIES,
                 pool_block=requests.adapters.DEFAULT_POOLBLOCK, burst_window=DEFAULT_BURST_WINDOW,
                 wait_window=DEFAULT_WAIT_WINDOW):
        self.throttle = BurstThrottle(pool_maxsize, burst_window, wait_window)
        super(MyHttpAdapter, self).__init__(pool_connections=pool_connections, pool_maxsize=pool_maxsize,
                                            max_retries=max_retries, pool_block=pool_block)

    def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None):
        request_successful = False
        response = None
        while not request_successful:
            wait_time = self.throttle.throttle()
            while wait_time > datetime.timedelta(0):
                gevent.sleep(wait_time.total_seconds(), ref=True)
                wait_time = self.throttle.throttle()

            response = super(MyHttpAdapter, self).send(request, stream=stream, timeout=timeout,
                                                       verify=verify, cert=cert, proxies=proxies)

            if response.status_code != 429:
                request_successful = True

        return response

设置：

requests_adapter = adapter.MyHttpAdapter(
    pool_connections=__CONCURRENT_LIMIT__,
    pool_maxsize=__CONCURRENT_LIMIT__,
    max_retries=0,
    pool_block=False,
    burst_window=datetime.timedelta(seconds=5),
    wait_window=datetime.timedelta(seconds=20))

requests_session = requests.session()
requests_session.mount('http://', requests_adapter)
requests_session.mount('https://', requests_adapter)

unsent_requests = (grequests.get(url,
                                 hooks={'response': handle_response},
                                 session=requests_session) for url in urls)
grequests.map(unsent_requests, size=__CONCURRENT_LIMIT__)

【讨论】：

【解决方案2】：

看看这个自动请求限制： https://pypi.python.org/pypi/RequestsThrottler/0.2.2

您可以设置每个请求之间的固定延迟量或设置在固定秒数内发送的请求数量（这基本上是相同的）：

import requests
from requests_throttler import BaseThrottler

request = requests.Request(method='GET', url='http://www.google.com')
reqs = [request for i in range(0, 5)]  # An example list of requests
with BaseThrottler(name='base-throttler', delay=1.5) as bt:
    throttled_requests = bt.multi_submit(reqs)

函数multi_submit 返回ThrottledRequest 的列表（参见文档：末尾的链接）。

然后您可以访问响应：

for tr in throttled_requests:
    print tr.response

或者，您可以通过指定在固定时间内发送的数量或请求（例如每 60 秒 15 个请求）来实现相同的目的：

import requests
from requests_throttler import BaseThrottler

request = requests.Request(method='GET', url='http://www.google.com')
reqs = [request for i in range(0, 5)]  # An example list of requests
with BaseThrottler(name='base-throttler', reqs_over_time=(15, 60)) as bt:
    throttled_requests = bt.multi_submit(reqs)

这两种解决方案都可以在不使用with 语句的情况下实现：

import requests
from requests_throttler import BaseThrottler

request = requests.Request(method='GET', url='http://www.google.com')
reqs = [request for i in range(0, 5)]  # An example list of requests
bt = BaseThrottler(name='base-throttler', delay=1.5)
bt.start()
throttled_requests = bt.multi_submit(reqs)
bt.shutdown()

【讨论】：

【解决方案3】：

看起来没有任何简单的机制来处理请求或 grequests 代码中的这个构建。唯一似乎存在的钩子是响应。

这是一个超级 hacky 解决方法，至少可以证明它是可能的 - 我修改了 grequests 以保留发出请求的时间列表，并休眠 AsyncRequest 的创建，直到每秒的请求数低于最大值。

class AsyncRequest(object):
    def __init__(self, method, url, **kwargs):
        print self,'init'
        waiting=True
        while waiting:
            if len([x for x in q if x > time.time()-15]) < 8:
                q.append(time.time())
                waiting=False
            else:
                print self,'snoozing'
                gevent.sleep(1)

您可以使用 grequests.imap() 以交互方式观看此内容

import time
import rg

urls = [
        'http://www.heroku.com',
        'http://python-tablib.org',
        'http://httpbin.org',
        'http://python-requests.org',
        'http://kennethreitz.com',
        'http://www.cnn.com',
]

def print_url(r, *args, **kwargs):
        print(r.url),time.time()

hook_dict=dict(response=print_url)
rs = (rg.get(u, hooks=hook_dict) for u in urls)
for r in rg.imap(rs):
        print r

我希望有一个更优雅的解决方案，但到目前为止我找不到。在会话和适配器中环顾四周。也许可以增强池管理器？

另外，我不会将此代码投入生产 - 'q' 列表永远不会被修剪，最终会变得相当大。另外，我不知道它是否真的像宣传的那样工作。当我查看控制台输出时，它看起来就是这样。

呃。只要看看这段代码，我就知道现在是凌晨 3 点。该上床睡觉了。

【讨论】：

【解决方案4】：

我遇到了类似的问题。这是我的解决方案。在你的情况下，我会这样做：

def worker():
    with rate_limit('slow.domain.com', 2):
        response = requests.get('https://slow.domain.com/path')
        text = response.text
    # Use `text`

假设您有多个要从中剔除的域，我会设置一个字典映射(domain, delay)，这样您就不会达到您的速率限制。

此代码假定您将使用 gevent 和猴子补丁。

from contextlib import contextmanager
from gevent.event import Event
from gevent.queue import Queue
from time import time


def rate_limit(resource, delay, _queues={}):
    """Delay use of `resource` until after `delay` seconds have passed.

    Example usage:

    def worker():
        with rate_limit('foo.bar.com', 1):
            response = requests.get('https://foo.bar.com/path')
            text = response.text
        # use `text`

    This will serialize and delay requests from multiple workers for resource
    'foo.bar.com' by 1 second.

    """

    if resource not in _queues:
        queue = Queue()
        gevent.spawn(_watch, queue)
        _queues[resource] = queue

    return _resource_manager(_queues[resource], delay)


def _watch(queue):
    "Watch `queue` and wake event listeners after delay."

    last = 0

    while True:
        event, delay = queue.get()

        now = time()

        if (now - last) < delay:
            gevent.sleep(delay - (now - last))

        event.set()   # Wake worker but keep control.
        event.clear()
        event.wait()  # Yield control until woken.

        last = time()


@contextmanager
def _resource_manager(queue, delay):
    "`with` statement support for `rate_limit`."

    event = Event()
    queue.put((event, delay))

    event.wait() # Wait for queue watcher to wake us.

    yield

    event.set()  # Wake queue watcher.

【讨论】：

【解决方案5】：

想要更好地控制节流/速率限制的人可以使用以下 Python 库：速率限制 2.2.1：https://pypi.org/project/ratelimit/

ratelimit 2.2.1：这个包引入了一个函数装饰器，防止函数被调用的频率超过 API 提供者所允许的频率。这应该可以防止 API 提供商通过遵守其速率限制来禁止您的应用程序。

from ratelimit import limits

import requests

FIFTEEN_MINUTES = 900

@limits(calls=15, period=FIFTEEN_MINUTES)
def call_api(url):
    response = requests.get(url)

    if response.status_code != 200:
        raise Exception('API response: {}'.format(response.status_code))
    return response

【讨论】：