背景

調(diào)用第三方接口時，常見的問題就是調(diào)用頻率過快，從而導致一系列的問題：可能會被封IP；可能會被封號；也可能會被限流等問題。

解決思路

最簡單的方法：限制一個時間段內(nèi)的調(diào)用頻率

ratelimit實現(xiàn)控制調(diào)用API的頻率

例如以下的測試代碼，使用多線程模擬請求調(diào)用test2()函數(shù)

import time
import threading

def test2():
    time.sleep(3)
    print('調(diào)用了函數(shù)')


def run_func():
    print('模擬普通請求...')
    test2()


if __name__ == '__main__':
    all_t = []
    for i in range(10):
        t = threading.Thread(target=run_func)
        all_t.append(t)

    for n in all_t:
        n.start()

    for k in all_t:
        k.join()

上面測試代碼中，啟動的10個線程中在同一個時刻調(diào)用了test2()函數(shù)，可以理解為1s內(nèi)請求API10次。下面需要引入ratelimit庫控制調(diào)用的頻率。

import time
from ratelimit import limits

@limits(calls=1, period=1)
def test2():
    time.sleep(3)
    print('調(diào)用了函數(shù)')

默認情況下，在15min內(nèi)允許15次請求，超出的請求會被丟棄。在這里設(shè)置calls=1,period=1，表示在1s內(nèi)只允許請求1次。

但是，運行上面代碼時，發(fā)生了其中一部分線程發(fā)生了ratelimit.exception.RateLimitException: too many calls錯誤，這是為什么呢？

增加阻塞等待

剛剛上面提到，在設(shè)置的時間內(nèi)，請求次數(shù)超出設(shè)置的值時，其他請求請求會被丟棄。因此，需要處理返回RateLimitException錯誤的情況。

增加一個裝飾器，使得超出的請求阻塞

import time
from functools import wraps

from ratelimit import RateLimitException


def sleep_and_retry(func):
    @wraps(func)
    def wrapper(*args, **kargs):
        while True:
            try:
                return func(*args, **kargs)
            except RateLimitException as exception:
                time.sleep(exception.period_remaining)
    return wrapper

結(jié)果

增加上面的裝飾器之后，超出的請求也實現(xiàn)堵塞，并能正常相應請求。

完整代碼

import threading
import time
from functools import wraps

from ratelimit import limits, RateLimitException


def sleep_and_retry(func):
    @wraps(func)
    def wrapper(*args, **kargs):
        while True:
            try:
                return func(*args, **kargs)
            except RateLimitException as exception:
                time.sleep(exception.period_remaining)
    return wrapper


@sleep_and_retry
@limits(calls=1, period=1)
def test2():
    time.sleep(3)
    print('調(diào)用了函數(shù)')


def run_func():
    print('模擬普通請求...')
    test2()


if __name__ == '__main__':
    all_t = []
    for i in range(10):
        t = threading.Thread(target=run_func)
        all_t.append(t)

    for n in all_t:
        n.start()

    for k in all_t:
        k.join()