1. import queue
  2. import logging
  3. import time
  4. import traceback
  6. class TaskQueue:
  7. def __init__(self, max_workers=5):
  8. self.task_queue = queue.Queue()
  9. self.workers = []
  10. self.max_workers = max_workers
  12. def start(self):
  13. """Starts the worker threads."""
  14. for _ in range(self.max_workers):
  15. worker = threading.Thread(target=self._worker)
  16. self.workers.append(worker)
  17. worker.daemon = True # Allow the main thread to exit even if workers are running
  18. worker.start()
  20. def submit(self, task, *args, **kwargs):
  21. """Submits a task to the queue."""
  22. self.task_queue.put((task, args, kwargs))
  24. def _worker(self):
  25. """Worker thread function."""
  26. while True:
  27. try:
  28. task, args, kwargs = self.task_queue.get(timeout=1) # Get task with timeout
  29. try:
  30. task(*args, **kwargs) # Execute the task
  31. except Exception as e:
  32. logging.error(f"Task failed: {e}")
  33. traceback.print_exc()
  34. finally:
  35. self.task_queue.task_done() # Mark task as completed
  36. except queue.Empty:
  37. # Timeout occurred, check if we should stop
  38. if all(not w.is_alive() for w in self.workers):
  39. break
  41. def wait_completion(self):
  42. """Waits for all tasks in the queue to be completed."""
  43. self.task_queue.join()
  45. import threading
  46. logging.basicConfig(level=logging.ERROR) #Configure basic logging
  48. if __name__ == '__main__':
  49. # Example usage
  50. task_queue = TaskQueue(max_workers=3)
  51. task_queue.start()
  53. def my_task(task_id):
  54. """A short-lived task."""
  55. print(f"Task {task_id} started")
  56. time.sleep(2)
  57. print(f"Task {task_id} completed")
  60. for i in range(10):
  61. task_queue.submit(my_task, i)
  63. task_queue.wait_completion()
  64. print("All tasks completed.")

Add your comment