1. import time
  2. import logging
  3. import queue
  5. # Configure logging
  6. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  8. class MessageQueueCompressor:
  9. def __init__(self, queue_name, batch_size=100, compression_interval=5):
  10. """
  11. Initializes the MessageQueueCompressor.
  13. Args:
  14. queue_name (str): The name of the message queue.
  15. batch_size (int): The number of messages to batch for compression.
  16. compression_interval (int): Time (seconds) between compression attempts.
  17. """
  18. self.queue_name = queue_name
  19. self.batch_size = batch_size
  20. self.compression_interval = compression_interval
  21. self.queue = queue.Queue() # Simulate message queue
  22. self.is_running = False
  24. def enqueue(self, message):
  25. """
  26. Adds a message to the queue.
  28. Args:
  29. message (any): The message to add.
  30. """
  31. self.queue.put(message)
  32. logging.info(f"Message enqueued: {message}")
  34. def compress(self):
  35. """
  36. Compresses messages in the queue.
  37. """
  38. if self.queue.empty():
  39. logging.info("Queue is empty. No compression needed.")
  40. return
  42. try:
  43. messages = []
  44. for _ in range(self.batch_size):
  45. messages.append(self.queue.get())
  47. if messages:
  48. # Simulate compression (e.g., serialize to JSON)
  49. compressed_messages = [str(msg) for msg in messages] # Simple string conversion
  50. logging.info(f"Compressed {len(messages)} messages.")
  52. # Simulate putting compressed messages back into the queue
  53. for compressed_msg in compressed_messages:
  54. self.enqueue(compressed_msg)
  55. else:
  56. logging.info("No messages to compress in this batch.")
  58. except queue.Empty:
  59. logging.warning("Queue is empty during compression.")
  60. except Exception as e:
  61. logging.error(f"Error during compression: {e}")
  63. def run(self):
  64. """
  65. Starts the compression process in a loop.
  66. """
  67. self.is_running = True
  68. while self.is_running:
  69. self.compress()
  70. time.sleep(self.compression_interval)
  72. def stop(self):
  73. """
  74. Stops the compression process.
  75. """
  76. self.is_running = False
  77. logging.info("Compression stopped.")
  79. if __name__ == '__main__':
  80. # Example usage
  81. compressor = MessageQueueCompressor("my_queue", batch_size=5, compression_interval=2)
  83. # Simulate adding messages to the queue
  84. for i in range(20):
  85. compressor.enqueue(f"Message {i}")
  86. time.sleep(0.5)
  88. # Start the compression process
  89. compressor.run()
  91. # Let it run for a while (e.g., 10 seconds)
  92. time.sleep(10)
  94. # Stop the compression process
  95. compressor.stop()

Add your comment