1. import time
  2. import threading
  3. import queue
  5. class TaskQueueCleaner:
  6. def __init__(self, queue, timeout=5):
  7. self.queue = queue
  8. self.timeout = timeout
  9. self.lock = threading.Lock() # For thread safety
  11. def clean_queue(self):
  12. """
  13. Cleans the queue by removing expired tasks.
  14. """
  15. with self.lock: # Acquire lock for thread safety
  16. while True:
  17. try:
  18. item = self.queue.get(timeout=self.timeout) # Get item with timeout
  19. # Process the item (e.g., check if it's expired)
  20. if self._is_expired(item):
  21. print(f"Removing expired task: {item}")
  22. self.queue.task_done() # Signal task completion
  23. else:
  24. self.queue.task_done() # Signal task completion
  25. except queue.Empty:
  26. # Queue is empty, exit the loop
  27. break
  29. def _is_expired(self, item):
  30. """
  31. Checks if a task is expired based on its expiration time.
  32. This is a placeholder; replace with your actual expiration logic.
  33. """
  34. # Replace with your logic to check task expiration
  35. if isinstance(item, dict) and 'expiration_time' in item:
  36. if time.time() > item['expiration_time']:
  37. return True
  38. return False
  40. if __name__ == '__main__':
  41. # Example Usage
  42. task_queue = queue.Queue()
  44. # Add some tasks with expiration times
  45. task_queue.put({'data': 'task1', 'expiration_time': time.time() + 2})
  46. task_queue.put({'data': 'task2', 'expiration_time': time.time() + 5})
  47. task_queue.put({'data': 'task3', 'expiration_time': time.time() + 1})
  48. task_queue.put({'data': 'task4', 'expiration_time': time.time() + 8})
  52. # Create a cleaner thread
  53. cleaner = TaskQueueCleaner(task_queue, timeout=1)
  54. cleaner_thread = threading.Thread(target=cleaner.clean_queue)
  55. cleaner_thread.daemon = True # Allow the main thread to exit even if the cleaner is running
  56. cleaner_thread.start()
  58. # Simulate adding new tasks
  59. time.sleep(2)
  60. task_queue.put({'data': 'task5', 'expiration_time': time.time() + 3})
  61. time.sleep(1)
  62. task_queue.put({'data': 'task6', 'expiration_time': time.time() + 6})
  65. # Wait for all tasks to be processed
  66. task_queue.join()
  67. print("All tasks processed.")

Add your comment