多Token轮询调用问题求助：如何充分利用全部Token

业务场景 #

• 真实环境有约100个IP、10个Token（示例用28个Token）
• 每个Token每分钟最多调用4次（实际需间隔至少25秒）
• 要求：每个IP仅用一个Token发起1次GET请求，所有Token需被充分利用

现有代码问题 #

原代码遍历Token时，找到第一个可用Token就直接break，导致前几个Token被重复复用，后续Token无法被使用；通过调整sleep(5)的方式适配性差，新增Token时需重新调整时长。

原代码：

import datetime
import requests
from time import sleep

tokens = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z']
dbs = [{'token': x, 'time': None} for x in tokens]
# 假设clients是已定义的IP列表
client_ips = [{'ip': x['ip'], 'desc': x['desc']} for x in clients]

thirty_days_ago = (datetime.datetime.now() - datetime.timedelta(days=30)).strftime('%Y-%m-%d')
requests_response_list = []
checked_ips = set()
for ip in client_ips:
    for db in dbs:
        now = datetime.datetime.now()
        if ip['ip'] in checked_ips:
            continue
        if db['time']:
            elapsed_time = (now - db['time']).total_seconds()
            if elapsed_time < 25:
                continue
        print(f"IP is {ip['ip']}, Token: {db['token']}.")
        response = requests.get(
            url=f"http://my.url.com/ip_addresses/{ip['ip']}",
            headers={
                'accept': 'application/json',
                'x-apikey': db['token'],
            }
        ).json()['data']['attributes']
        for report in response:
            # report['date'] 是时间戳，转换为Y-m-d格式
            date = datetime.datetime.fromtimestamp(report['date']).strftime('%Y-%m-%d')
            if thirty_days_ago <= date:
                requests_response_list.append({
                    'ip': ip['ip'],
                    'date': date,
                })
        db['time'] = now
        checked_ips.add(ip['ip'])
        sleep(5)
        break

优化方案 #

核心思路：改用轮询机制分配Token，确保每个Token被依次使用；动态检查Token冷却时间，无需固定休眠时长。

优化后的代码：

import datetime
import requests
from time import sleep

tokens = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z']
# 用字典记录每个Token的最后使用时间，初始为None
token_last_used = {token: None for token in tokens}
# 假设clients是已定义的IP列表
client_ips = [{'ip': x['ip'], 'desc': x['desc']} for x in clients]

thirty_days_ago = (datetime.datetime.now() - datetime.timedelta(days=30)).strftime('%Y-%m-%d')
requests_response_list = []
checked_ips = set()
token_index = 0  # 轮询索引，记录当前尝试的Token位置

for ip in client_ips:
    ip_addr = ip['ip']
    if ip_addr in checked_ips:
        continue
    
    while True:
        current_token = tokens[token_index]
        now = datetime.datetime.now()
        last_used = token_last_used[current_token]
        
        # 检查Token是否可用：未使用过 或 已满足间隔要求
        if not last_used or (now - last_used).total_seconds() >= 25:
            # 发起请求
            print(f"IP is {ip_addr}, Token: {current_token}.")
            response = requests.get(
                url=f"http://my.url.com/ip_addresses/{ip_addr}",
                headers={
                    'accept': 'application/json',
                    'x-apikey': current_token,
                }
            ).json()['data']['attributes']
            
            # 处理响应数据
            for report in response:
                date = datetime.datetime.fromtimestamp(report['date']).strftime('%Y-%m-%d')
                if thirty_days_ago <= date:
                    requests_response_list.append({
                        'ip': ip_addr,
                        'date': date,
                    })
            
            # 更新Token最后使用时间
            token_last_used[current_token] = now
            checked_ips.add(ip_addr)
            
            # 更新轮询索引，循环遍历所有Token
            token_index = (token_index + 1) % len(tokens)
            break
        else:
            # 当前Token不可用，尝试下一个
            token_index = (token_index + 1) % len(tokens)
            # 短暂休眠避免空循环占用资源
            sleep(0.1)

优化说明 #

1. 轮询分配Token：通过token_index循环遍历所有Token，确保每个Token都能被依次使用，不会出现前几个Token被重复占用的情况
2. 动态检查冷却时间：每次尝试使用Token时，先判断是否满足间隔要求，无需固定sleep时长，新增Token时无需调整逻辑
3. 避免空循环浪费资源：当Token都处于冷却时，短暂休眠0.1秒后再尝试，降低CPU占用

内容的提问来源于stack exchange，提问作者Saeed