Verwenden von Python-Threads, um tausende Aufrufe an eine langsame API mit einem Ratenlimit zu senden

Question

Ich möchte tausende Aufrufe an eine API senden, die langsam ist - einige zehn Sekunden, um eine Antwort zu erhalten. Die einzige Grenze ist, dass ich höchstens eine Anfrage pro Sekunde machen kann. Was ist der beste Weg, dies zu tun? Ich denke, der folgende Code funktioniert, aber ich denke, ich sollte die Threading-Bibliothek irgendwie besser nutzen können. Ich verwende Python 3,3

last_job = datetime.now() 
for work in work_list: 
    while (datetime.now()-last_job).total_seconds() < 1 or threading.active_count() >= max_threads: 
     time.sleep(.1) 
    threading.Thread(target=work_function, args=[work]).start() 
    last_job = datetime.now() 

Answer 1

Wenn Sie eine Reihe von Jobs mit einer festen Größe Thread-Pool ausführen möchten, können Sie concurrent.futures.ThreadPoolExecutor wie folgt verwenden:

from concurrent.futures import ThreadPoolExecutor 
with ThreadPoolExecutor(max_workers=5) as executor: 
    for work in work_list: 
     executor.submit(work_function, work) 

Wenn Sie wollen sicherstellen, dass du höchstens einen API-Aufruf pro Sekunde machst, dann musst du das aus deinem work_function heraus tun. Sie können dies nicht tun, wenn Sie den Job senden, weil Sie nicht wissen, wie lange der Job in der Warteschlange wartet, bis ein Thread verfügbar wird.

Wenn es nach mir ginge, ich die Raten-Code in die eigene Klasse zu begrenzen setzen würde, so dass sie wiederverwendbar ist:

from collections import Iterator 
from threading import Lock 
import time 

class RateLimiter(Iterator): 
    """Iterator that yields a value at most once every 'interval' seconds.""" 
    def __init__(self, interval): 
     self.lock = Lock() 
     self.interval = interval 
     self.next_yield = 0 

    def __next__(self): 
     with self.lock: 
      t = time.monotonic() 
      if t < self.next_yield: 
       time.sleep(self.next_yield - t) 
       t = time.monotonic() 
      self.next_yield = t + self.interval 

api_rate_limiter = RateLimiter(1) 

def work_function(work): 
    next(api_rate_limiter) 
    call_api(...) 

time.monotonic in Python 3.3 eingeführt wurde; in älteren Versionen von Python können Sie time.time verwenden, aber dies kann rückwärts springen, wenn die Systemuhr Änderungen, so müssen Sie sicherstellen, dass diese nicht überlang schläft verursacht:

   time.sleep(min(self.next_yield - t, self.interval))