2

我有一个大约 300K URL 的列表，用于我需要从中获取数据的 API。

API 限制为每秒 100 次调用。

我已经为异步创建了一个类，但这正在快速运行，并且我在 API 上遇到错误。

如何减慢异步速度，以便每秒可以进行 100 次调用？

import grequests

lst = ['url.com','url2.com']

class Test:
    def __init__(self):
        self.urls = lst

    def exception(self, request, exception):
        print ("Problem: {}: {}".format(request.url, exception))

    def async(self):
        return grequests.map((grequests.get(u) for u in self.urls), exception_handler=self.exception, size=5)



    def collate_responses(self, results):
        return [x.text for x in results]

test = Test()
#here we collect the results returned by the async function
results = test.async()
response_text = test.collate_responses(results)
4

3 回答 3

2

我采取的第一步是创建一个对象，该对象每 t ms 最多可以分发 n 个硬币。

import time

class CoinsDistribution:
    """Object that distribute a maximum of maxCoins every timeLimit ms"""
    def __init__(self, maxCoins, timeLimit):
        self.maxCoins = maxCoins
        self.timeLimit = timeLimit
        self.coin = maxCoins
        self.time = time.perf_counter()


    def getCoin(self):
        if self.coin <= 0 and not self.restock():
            return False

        self.coin -= 1
        return True

    def restock(self):
        t = time.perf_counter()
        if (t - self.time) * 1000 < self.timeLimit:
            return False
        self.coin = self.maxCoins
        self.time = t
        return True

现在我们需要一种强制函数只有在他们可以获得硬币时才被调用的方法。为此，我们可以编写一个可以这样使用的装饰器函数：

@limitCalls(callLimit=1, timeLimit=1000)
def uniqFunctionRequestingServer1():
    return 'response from s1'

但有时，多个函数调用请求同一个服务器，所以我们希望它们从同一个 CoinsDistribution 对象中获取硬币。因此，装饰器的另一种用途是提供 CoinsDistribution 对象：

server_2_limit = CoinsDistribution(3, 1000)

@limitCalls(server_2_limit)
def sendRequestToServer2():
    return 'it worked !!'

@limitCalls(server_2_limit)
def sendAnOtherRequestToServer2():
    return 'it worked too !!'

我们现在必须创建装饰器，它可以采用 CoinsDistribution 对象或足够的数据来创建新的。

import functools

def limitCalls(obj=None, *, callLimit=100, timeLimit=1000):
    if obj is None:
        obj = CoinsDistribution(callLimit, timeLimit)

    def limit_decorator(func):
        @functools.wraps(func)
        def limit_wrapper(*args, **kwargs):
            if obj.getCoin():
                return func(*args, **kwargs)
            return 'limit reached, please wait'
        return limit_wrapper
    return limit_decorator

它完成了！现在，您可以限制您使用的任何 API 的调用次数，并且如果您必须管理大量 CoinsDistribution 对象（针对不同的 API 端点或不同的 API），您可以构建一个字典来跟踪您的 CoinsDistribution 对象。

注意：这里我选择在没有可用硬币的情况下返回错误消息。您应该根据您的需要调整此行为。

于 2018-09-02T00:50:47.133 回答
1

您可以跟踪已经过去了多少时间并决定是否要执行更多请求。

这将每秒打印 100 个数字，例如：

from datetime import datetime
import time

start = datetime.now()
time.sleep(1);
counter = 0
while (True):
    end = datetime.now()
    s = (end-start).seconds
    if (counter >= 100):
        if (s <= 1):
            time.sleep(1) # You can keep track of the time and sleep less, actually
            start = datetime.now()
            counter = 0
    print(counter)
    counter += 1
于 2018-08-24T13:11:17.397 回答
-2

SO中的另一个问题确切地显示了如何做到这一点。顺便说一句，您需要的通常称为throttling。

于 2018-08-24T12:52:21.537 回答