1. import queue
  2. import threading
  3. import time
  5. class MessageQueue:
  6. def __init__(self, max_size=100):
  7. """
  8. Initializes a message queue with a maximum size.
  9. """
  10. self.queue = queue.Queue(maxsize=max_size) # Use queue.Queue for thread safety
  11. self.lock = threading.Lock() # For additional thread safety if needed
  13. def put(self, item, block=True, timeout=None):
  14. """
  15. Adds an item to the queue. Blocks if the queue is full.
  16. """
  17. self.queue.put(item, block=block, timeout=timeout) # Use queue.put for blocking behavior
  19. def get(self, block=True, timeout=None):
  20. """
  21. Removes and returns an item from the queue. Blocks if the queue is empty.
  22. """
  23. return self.queue.get(block=block, timeout=timeout) # Use queue.get for blocking behavior
  25. def size(self):
  26. """
  27. Returns the current number of items in the queue.
  28. """
  29. with self.lock: #Protect queue size from race conditions if needed
  30. return self.queue.qsize()
  32. def is_empty(self):
  33. """
  34. Returns True if the queue is empty, False otherwise.
  35. """
  36. return self.queue.empty()
  38. def is_full(self):
  39. """
  40. Returns True if the queue is full, False otherwise.
  41. """
  42. return self.queue.full()
  45. def enforce_queue_limit(queue_instance, item):
  46. """
  47. Example function to demonstrate queue limit enforcement.
  48. """
  49. if queue_instance.size() >= 95: #limit is 100, allow 5% buffer
  50. print(f"Queue is almost full. Dropping item: {item}")
  51. return False # Indicate item was dropped
  52. else:
  53. queue_instance.put(item) # Add the item to the queue
  54. print(f"Added item to queue: {item}. Queue size: {queue_instance.size()}")
  55. return True # Indicate item was added
  58. if __name__ == '__main__':
  59. my_queue = MessageQueue(max_size=100) # Create a queue with a maximum size of 100
  61. # Simulate adding items to the queue
  62. for i in range(120):
  63. enforce_queue_limit(my_queue, i)
  64. time.sleep(0.01) # Simulate some processing time
  66. print("Queue is now full.")
  67. #Example of trying to add more items after queue is full
  68. for i in range(120,130):
  69. enforce_queue_limit(my_queue, i)
  70. time.sleep(0.01)

Add your comment