python - 速率限制api多进程

Question

我有一个非常简单的代码，如果 id 来自文件，我会加载一个列表，然后遍历列表中的每个 id 并调用一个 api，在那里我传递 id 值并将 api 响应内容转储到一个文件中。
我想通过并行 api 调用来加快这个过程，但是 api 服务器每秒最多只允许 5 个调用。另一个关键考虑因素是 api pull 很慢，平均每次调用需要 10 秒才能完成。
我希望能够拥有多个并行进程，这些进程可以通过某种方式确保在一秒钟内最多发生 5 个调用。
这是当前的代码:

import pandas as pd
import numpy as np
from joblib import Parallel, delayed

ids = pd.read_csv('data.csv')

ids = ids['Id'].values.tolist()

def dump_data(df,idx):

    filename = base_dir+'\\'+str(idx)+'.csv'
    data.to_csv(filename, header= True, index=False) #write data to file

def get_api(idx):

   data = call_some_api(idx)  #api returns data as pandas dataframe, take about 10 secs
   dump_data(df,idx)


Parallel(n_jobs=10, verbose = 50)(delayed(get_api)(idx) for idx in ids) 

我目前正在使用 joblib，但如果此解决方案有更好的库，则可以改用它。
我如何确保在任何给定的时间内发出的请求不会超过 5 个？ (同时他尽可能快地完成所有请求)
另外我在 Windows 上使用 Python 3.9

Answer 1

可以使用 concurrent.futures 库和一次运行 5 个线程的循环，我将工作线程的数量限制为 50，但这可能会更高，因为线程没有执行任何 CPU 密集型任务。
使用最后一个 sleep(1) 语句，您可以保证每秒绕过或不到 5 次调用(因为循环处理时间)

import pandas as pd
import numpy as np
import concurrent.futures, time

ids = pd.read_csv('data.csv')

ids = ids['Id'].values.tolist()

def dump_data(df,idx):
    filename = base_dir+'\\'+str(idx)+'.csv'
    data.to_csv(filename, header= True, index=False) #write data to file

def get_api(idx):
    data = call_some_api(idx)  #api returns data as pandas dataframe, take about 10 secs
    dump_data(df,idx)

N = 5 # number of requests per second
L = 10 # latency per request

with concurrent.futures.ThreadPoolExecutor(max_workers=N*L) as ex:
    # reshape IDs into a list of lists of N items each and loop over them
    for id_list in ((ids[i] for i in range(j*N,min(j*N+N,len(ids))) ) for j in range(round(len(ids)/N+0.5)) )

        # wait if there are items on the work queue 
        while ex._work_queue.qsize()>0: time.sleep(1)   

        ex.map(get_api,id_list)
        time.sleep(1)