电报机器人的 API 速率限制

Question

我正在开发一个带有模块 pyTelegramBotAPI 的机器人，它通过 webhook 工作，并安装了 Flask+Gunicorn 作为 webhook 的服务器。 Gunicorn 正在与 5 名工人合作以提高速度，我的项目结构如下：

app.py
bot.py

在 bot.py 中，我有一个处理更新的函数：

def process_the_update(update):
    logger.info(update)
    update = types.Update.de_json(update)
    bot.process_new_updates([update])

我在 app.py 中导入了这个函数，所以，每当有更新到来时，app.py 都会调用这个函数，而 bot 会处理更新。在我的机器人中，用户可以调用一个命令，该命令将使用外部 api 来获取一些信息。问题是，这个外部 api 每秒有 3 个请求的限制。我需要配置一个具有这种速率限制的机器人。首先我想用 Queue 来做这件事，代码如下：

lock_queue = Queue(1)
requests_queue = Queue(3)
def api_request(argument):
    if lock_queue.empty():
        try:
            requests_queue.put_nowait(time.time())
        except queue.Full:
            lock_queue.put(1)
            first_request_time = requests_queue.get()
            logger.info('First request time: ' + str(first_request_time))
            current_time = time.time()
            passed_time = current_time - first_request_time
            if passed_time >= 1:
                requests_queue.put_nowait(time.time())
                lock_queue.get()
            else:
                logger.info(passed_time)
                time.sleep(1 - passed_time)
                requests_queue.put_nowait(time.time())
                lock_queue.get()
    else:
        lock_queue.put(1)
        first_request_time = vk_requests_queue.get()
        logger.info('First request time: ' + str(first_request_time))
        current_time = time.time()
        passed_time = current_time - first_request_time
        if passed_time >= 1:
            requests_queue.put_nowait(time.time())
            lock_queue.get()
        else:
            logger.info(passed_time)
            time.sleep(1 - passed_time)
            requests_queue.put_nowait(time.time())
            lock_queue.get()
    result = make_api_request(argument) # requests are made too by external module.
    return result 

逻辑是，正如我所想，因为模块 pyTelegramBotAPI 使用线程来更快地处理更新，所有线程都会检查 requests_queue，这将有 3 个最后一个 api_requests 的时间，因此发出的 3 个请求中的第一个的时间将是与当前时间相比（检查，如果一秒钟过去了）。而且，因为我需要确定只有一个线程会同时进行这种比较，所以我创建了 lock_queue。 但是，问题在于，首先，gunicorn 使用 5 个 worker，所以总是有可能，来自用户的所有消息都将在不同的进程中处理，并且这些进程会有自己的队列。其次，即使我将工作人员的数量设置为默认值（1 个工作人员），我仍然会收到 429 错误，所以我认为我的代码根本无法按我的意愿工作。

我想用 redis 做速率限制，所以每次在每个线程和进程中，bot 都会检查最后 3 个请求的时间，但我仍然不确定，这是正确的方法，我不确定，这个怎么写。

如果有人提出任何想法甚至代码示例（外部 api 不提供任何 x-rate-limit 标头），我会很高兴

Answer 1

写了这个函数，使用redis统计请求（基于这个https://www.binpress.com/tutorial/introduction-to-rate-limiting-with-redis/155教程）

import redis

r_db = redis.Redis(port=port, db=db)

def limit_request(request_to_make, limit=3, per=1, request_name='test', **kwargs):
    over_limit_lua_ = '''
    local key_name = KEYS[1]
    local limit = tonumber(ARGV[1])
    local duration = ARGV[2]

    local key = key_name .. '_num_of_requests'
    local count = redis.call('INCR', key)
    if tonumber(count) > limit then
        local time_left = redis.call('PTTL', key)
        return time_left
    end
    redis.call('EXPIRE', key, duration)
    return -2
    '''

    if not hasattr(r_db, 'over_limit_lua'):
        r_db.over_limit_lua = r_db.register_script(over_limit_lua_)

    request_possibility = int(r_db.over_limit_lua(keys=request_name, args=[limit, per]))
    if request_possibility > 0:
        time.sleep(request_possibility / 1000.0)
        return limit_request(request_to_make, limit, per, request_name, **kwargs)
    else:
        request_result = request_to_make(**kwargs)
        return request_result