如何为域名列表的每个请求分配不同API Key并并发调用？

要实现为每个域名请求分配不同API密钥并并发调用，我们可以通过轮询API密钥列表的方式分配密钥，同时保留原有的异步调用逻辑。以下是修改后的代码及关键说明：

关键修改点 #

• 新增API密钥列表，通过索引取模实现轮询分配
• 优化异常处理，避免请求失败时直接静默跳过
• 保留原有的异步并发逻辑，同时可按需限制并发数（可选）

修改后的完整代码 #

import asyncio
import aiohttp
from time import perf_counter
import csv

path = "*******************"
domains = []
total_count = 0
# 定义你的API密钥列表
API_KEYS = ["API_KEY_1", "API_KEY_2", "API_KEY_3"]

with open(path, 'r') as file:
    csvreader = csv.reader(file)
    for row in csvreader:
        try:
            website = row[4].split("//")[-1].split("www.")[-1].split('/')[0]
            if website == "":
                continue
            domains.append(website)
        except Exception as e:
            print(f"处理域名时出错: {e}")
            continue
sample = domains[0:50]

async def fetch(session, body):
    global total_count
    try:
        async with session.post('https://****************', json=body) as r:
            if r.status != 200:
                print(f"请求失败，状态码: {r.status}，域名: {body['domain']}")
                return
            enrich_response = await r.json()
            employees = enrich_response.get('employees', [])
            for employee in employees:
                job_title = employee.get('job_title', '').strip()
                if job_title in ("Owner", "CEO"):
                    print(employee)
                    print("************************************************")
                    total_count += 1
                    print("Total Count:", total_count)
    except Exception as e:
        print(f"请求异常: {e}，域名: {body['domain']}")

async def fetch_all(session, bodies):
    tasks = []
    # 可选：限制并发数，比如同时最多10个请求，避免触发API限流
    semaphore = asyncio.Semaphore(10)
    async def bounded_fetch(body):
        async with semaphore:
            await fetch(session, body)
    for body in bodies:
        task = asyncio.create_task(bounded_fetch(body))
        tasks.append(task)
    await asyncio.gather(*tasks)

async def main():
    bodies = []
    for idx, domain in enumerate(sample):
        # 轮询分配API密钥：索引取模实现循环使用
        api_key = API_KEYS[idx % len(API_KEYS)]
        body = {
            "api_key": api_key,
            "domain": domain
        }
        bodies.append(body)
    
    async with aiohttp.ClientSession() as session:
        await fetch_all(session, bodies)

if __name__ == '__main__':
    start = perf_counter()
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"主程序异常: {e}")
    stop = perf_counter()
    print(f"总耗时: {stop - start:.2f} 秒")
    print(f"找到的Owner/CEO总数: {total_count}")

核心说明 #

1. 密钥轮询分配：通过idx % len(API_KEYS)让每个域名请求按顺序循环使用API密钥，确保3个密钥被均匀分配。
2. 异常处理优化：新增了对域名解析异常、请求失败、响应解析异常的捕获和打印，便于排查问题。
3. 可选并发限制：使用asyncio.Semaphore限制同时发起的请求数，避免因并发过高触发API的限流机制（可根据API的并发限制调整数值）。
4. 变量安全：原代码的total_count使用了全局变量，若需要更安全的多任务共享变量，可改用asyncio.Lock来保护计数操作（避免多任务同时修改导致计数错误）。

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