1. import time
  2. from collections import deque
  4. class TaskQueueThrottler:
  5. def __init__(self, max_requests, interval):
  6. """
  7. Initializes the throttler.
  9. Args:
  10. max_requests (int): The maximum number of requests allowed within the interval.
  11. interval (float): The time interval in seconds.
  12. """
  13. self.max_requests = max_requests
  14. self.interval = interval
  15. self.request_queue = deque() # Queue to store timestamps of requests
  16. self.lock = time.Lock()
  18. def is_allowed(self):
  19. """
  20. Checks if a request is allowed based on the throttle.
  22. Returns:
  23. bool: True if the request is allowed, False otherwise.
  24. """
  25. with self.lock:
  26. now = time.time()
  27. while self.request_queue and self.request_queue[0] <= now - self.interval:
  28. self.request_queue.popleft() # Remove expired requests
  30. if len(self.request_queue) < self.max_requests:
  31. self.request_queue.append(now)
  32. return True
  33. else:
  34. return False
  36. def submit(self):
  37. """
  38. Submits a request to the queue, throttling if necessary.
  39. """
  40. if self.is_allowed():
  41. # Simulate task execution (replace with your actual task)
  42. print("Executing task...")
  43. time.sleep(1) # Simulate task taking 1 second
  44. else:
  45. print("Throttled. Request denied.")
  47. #Add timestamp to request queue for throttling
  48. with self.lock:
  49. self.request_queue.append(time.time())
  51. time.sleep(self.interval) #Wait for the interval
  54. if __name__ == '__main__':
  55. throttler = TaskQueueThrottler(max_requests=3, interval=2) # Allow 3 requests every 2 seconds
  57. for i in range(5):
  58. if throttler.is_allowed():
  59. throttler.submit()
  60. else:
  61. print(f"Skipping request {i}")

Add your comment