{"id":165,"date":"2025-12-20T02:32:35","date_gmt":"2025-12-19T18:32:35","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=165"},"modified":"2025-12-20T02:32:35","modified_gmt":"2025-12-19T18:32:35","slug":"how-to-build-a-high-performance-distributed-task-routing-system-using-kombu-with-topic-exchanges-and-concurrent-workers","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=165","title":{"rendered":"How to Build a High-Performance Distributed Task Routing System Using Kombu with Topic Exchanges and Concurrent Workers"},"content":{"rendered":"

In this tutorial, we build a fully functional event-driven workflow using Kombu<\/strong><\/a>, treating messaging as a core architectural capability. We walk through step by step the setup of exchanges, routing keys, background workers, and concurrent producers, allowing us to observe a real distributed system. As we implement each component, we see how clean message flow, asynchronous processing, and routing patterns give us the same power that production microservices rely on every day. Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong><\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip install kombu\n\n\nimport threading\nimport time\nimport logging\nimport uuid\nimport datetime\nimport sys\n\n\nfrom kombu import Connection, Exchange, Queue, Producer, Consumer\nfrom kombu.mixins import ConsumerMixin\n\n\nlogging.basicConfig(\n   level=logging.INFO,\n   format='%(message)s',\n   handlers=[logging.StreamHandler(sys.stdout)],\n   force=True\n)\nlogger = logging.getLogger(__name__)\n\n\nBROKER_URL = \"memory:\/\/localhost\/\"<\/code><\/pre>\n<\/div>\n<\/div>\n
We begin by installing Kombu, importing dependencies, and configuring logging so we can clearly see every message flowing through the system. We also set the in-memory broker URL, allowing us to run everything locally in Colab without needing RabbitMQ. This setup forms the foundation for our distributed messaging workflow. Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong><\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
media_exchange = Exchange('media_exchange', type='topic', durable=True)\n\n\ntask_queues = [\n   Queue('video_queue', media_exchange, routing_key='video.#'),\n   Queue('audit_queue', media_exchange, routing_key='#'),\n]\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We define a topic exchange to flexibly route messages using wildcard patterns. We also create two queues: one dedicated to video-related tasks and another audit queue that listens to everything. Using topic routing, we can precisely control how messages flow across the system. Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong><\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class Worker(ConsumerMixin):\n   def __init__(self, connection, queues):\n       self.connection = connection\n       self.queues = queues\n       self.should_stop = False\n\n\n   def get_consumers(self, Consumer, channel):\n       return [\n           Consumer(queues=self.queues,\n                    callbacks=[self.on_message],\n                    accept=['json'],\n                    prefetch_count=1)\n       ]\n\n\n   def on_message(self, body, message):\n       routing_key = message.delivery_info['routing_key']\n       payload_id = body.get('id', 'unknown')\n\n\n       logger.info(f\"n\"\u26a1\" RECEIVED MSG via key: [{routing_key}]\")\n       logger.info(f\"   Payload ID: {payload_id}\")\n      \n       try:\n           if 'video' in routing_key:\n               self.process_video(body)\n           elif 'audit' in routing_key:\n               logger.info(\"   \"\ud83d\udd0d\" [Audit] Logging event...\")\n          \n           message.ack()\n           logger.info(f\"   \"\u2705\" ACKNOWLEDGED\")\n\n\n       except Exception as e:\n           logger.error(f\"   \"\u274c\" ERROR: {e}\")\n\n\n   def process_video(self, body):\n       logger.info(\"   \"\u2699\"  [Processor] Transcoding video (Simulating work...)\")\n       time.sleep(0.5)<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement a custom worker using Kombu\u2019s ConsumerMixin to run it in a background thread. In the message callback, we inspect the routing key, invoke the appropriate processing function, and acknowledge the message. This worker architecture gives us clean, concurrent message consumption with full control. Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong><\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def publish_messages(connection):\n   producer = Producer(connection)\n  \n   tasks = [\n       ('video.upload', {'file': 'movie.mp4'}),\n       ('user.login', {'user': 'admin'}),\n   ]\n\n\n   logger.info(\"n\"\ud83d\ude80\" PRODUCER: Starting to publish messages...\")\n  \n   for r_key, data in tasks:\n       data['id'] = str(uuid.uuid4())[:8]\n      \n       logger.info(f\"\"\ud83d\udce4\" SENDING: {r_key} -> {data}\")\n      \n       producer.publish(\n           data,\n           exchange=media_exchange,\n           routing_key=r_key,\n           serializer='json'\n       )\n       time.sleep(1.5)\n\n\n   logger.info(\"\"\ud83c\udfc1\" PRODUCER: Done.\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We now build a producer that sends structured JSON payloads into the exchange with different routing keys. We generate unique IDs for each event and observe how they are routed to other queues. This mirrors real-world microservice event publishing, where producers and consumers remain decoupled. Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong><\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def run_example():\n   with Connection(BROKER_URL) as conn:\n       worker = Worker(conn, task_queues)\n       worker_thread = threading.Thread(target=worker.run)\n       worker_thread.daemon = True\n       worker_thread.start()\n      \n       logger.info(\"\"\u2705\" SYSTEM: Worker thread started.\")\n       time.sleep(1)\n\n\n       try:\n           publish_messages(conn)\n           time.sleep(2)\n       except KeyboardInterrupt:\n           pass\n       finally:\n           worker.should_stop = True\n           logger.info(\"n\"\ud83d\udc4b\" SYSTEM: Execution complete.\")\n\n\nif __name__ == \"__main__\":\n   run_example()<\/code><\/pre>\n<\/div>\n<\/div>\n
We start the worker in a background thread and fire the producer in the main thread. This structure gives us a mini distributed system running in Colab. By observing the logs, we see messages published \u2192 routed \u2192 consumed \u2192 acknowledged, completing the full event-processing lifecycle.<\/p>\n
In conclusion, we orchestrated a dynamic, distributed task-routing pipeline that processes real-time events with clarity and precision. We witnessed how Kombu abstracts away the complexity of messaging systems while still giving us fine-grained control over routing, consumption, and worker concurrency. As we see messages move from producer to exchange to queue to worker, we gained a deeper appreciation for the elegance of event-driven system design, and we are now well-equipped to scale this foundation into robust microservices, background processors, and enterprise-grade workflows.<\/p>\n
\n
Check out the\u00a0FULL CODES<\/a>.<\/em><\/strong>\u00a0Feel free to check out our\u00a0GitHub Page for Tutorials, Codes and Notebooks<\/a><\/mark><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>.<\/p>\n
The post How to Build a High-Performance Distributed Task Routing System Using Kombu with Topic Exchanges and Concurrent Workers<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a f…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-165","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/165","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=165"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/165\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=165"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=165"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=165"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}