Simulate RabbitMQ QoS Prefetch Count in Python

import threading
import time
import queue


class RabbitMQMock:
    def __init__(self, prefetch_count=1):
        self.prefetch_count = prefetch_count
        self.channel_queue = queue.Queue()
        self.currently_processing = 0
        self.lock = threading.Lock()

    def start_consuming(self, messages, worker_count=3):
        """Simulate QoS: limit unacked messages per worker to prefetch_count."""
        for msg in messages:
            self.channel_queue.put(msg)

        workers = []
        for _ in range(worker_count):
            t = threading.Thread(target=self._worker, args=(messages,))
            workers.append(t)
            t.start()

        for t in workers:
            t.join()
        print(f"All messages processed. Prefetch count: {self.prefetch_count}")

    def _worker(self, messages):
        while True:
            with self.lock:
                if self.currently_processing >= self.prefetch_count:
                    continue
                try:
                    msg = self.channel_queue.get_nowait()
                except queue.Empty:
                    return
                self.currently_processing += 1

            print(f"Worker {threading.current_thread().name} processing: {msg}")
            time.sleep(0.1)  # simulate work

            with self.lock:
                self.currently_processing -= 1
                self.channel_queue.task_done()


if __name__ == "__main__":
    mock = RabbitMQMock(prefetch_count=2)
    mock.start_consuming(["msg1", "msg2", "msg3", "msg4", "msg5"], worker_count=3)