En este tutorial, creamos un flujo de trabajo basado en eventos completamente funcional utilizando Kombu, tratando la mensajería como una capacidad arquitectónica central. Recorremos paso a paso la configuración de intercambios, claves de enrutamiento, trabajadores en segundo plano y productores concurrentes, lo que nos permite observar un sistema distribuido real. A medida que implementamos cada componente, vemos cómo el flujo de mensajes limpio, el procesamiento asincrónico y los patrones de enrutamiento nos brindan el mismo poder del que dependen los microservicios de producción todos los días. Consulte los CÓDIGOS COMPLETOS.
Comenzamos instalando Kombu, importando dependencias y configurando el registro para que podamos ver claramente cada mensaje que fluye a través del sistema. También configuramos la URL del agente en memoria, lo que nos permite ejecutar todo localmente en Colab sin necesidad de RabbitMQ. Esta configuración constituye la base de nuestro flujo de trabajo de mensajería distribuida. Consulte los CÓDIGOS COMPLETOS.
Queue(‘video_queue’, media_exchange, routing_key=’video.#’),
Queue(‘audit_queue’, media_exchange, routing_key=’#’),
]
Definimos un intercambio de temas para enrutar mensajes de manera flexible utilizando patrones comodín. También creamos dos colas: una dedicada a tareas relacionadas con vídeos y otra cola de auditoría que escucha todo. Al utilizar el enrutamiento de temas, podemos controlar con precisión cómo fluyen los mensajes a través del sistema. Consulte los CÓDIGOS COMPLETOS.
Consumer(queues=self.queues,
callbacks=[self.on_message]aceptar =[‘json’]prefetch_count=1) ]def on_message(self, body, message): clave_enrutamiento = mensaje.entrega_info[‘routing_key’]
payload_id = body.get(‘id’, ‘unknown’) logger.info(f”\n⚡ RECIBIDO MSG mediante clave: [{routing_key}]”) logger.info(f” ID de carga útil: {payload_id}”) intente: si ‘video’ en enrutamiento_key: self.process_video(body) elif ‘audit’ en enrutamiento_key: logger.info(” 🔍 [Audit] Evento de registro…”) message.ack() logger.info(f” ✅ RECONOCIDO”) excepto Excepción como e: logger.error(f” ❌ ERROR: {e}”) def process_video(self, body): logger.info(” ⚙️ [Processor] Transcodificación de vídeo (Simulación de trabajo…)”) time.sleep(0.5)
Implementamos un trabajador personalizado usando ConsumerMixin de Kombu para ejecutarlo en un hilo en segundo plano. En la devolución de llamada del mensaje, inspeccionamos la clave de enrutamiento, invocamos la función de procesamiento adecuada y confirmamos el mensaje. Esta arquitectura de trabajo nos brinda un consumo de mensajes limpio y simultáneo con control total. Consulte los CÓDIGOS COMPLETOS.
(‘video.upload’, {‘file’: ‘movie.mp4’}),
(‘user.login’, {‘user’: ‘admin’}),
]
logger.info(“\n🚀 PRODUCTOR: Comenzando a publicar mensajes…”) para r_key, datos en tareas: datos[‘id’] = str(uuid.uuid4())[:8]
logger.info(f”📤 ENVÍO: {r_key} -> {data}”) productor.publish( data, exchange=media_exchange, route_key=r_key, serializer=”json” ) time.sleep(1.5) logger.info(“🏁 PRODUCTOR: Listo.”)
Ahora creamos un productor que envía cargas útiles JSON estructuradas al intercambio con diferentes claves de enrutamiento. Generamos identificaciones únicas para cada evento y observamos cómo se enrutan a otras colas. Esto refleja la publicación de eventos de microservicios en el mundo real, donde los productores y consumidores permanecen desacoplados. Consulte los CÓDIGOS COMPLETOS.
Iniciamos al trabajador en un hilo en segundo plano y despedimos al productor en el hilo principal. Esta estructura nos proporciona un mini sistema distribuido que se ejecuta en Colab. Al observar los registros, vemos mensajes publicados → enrutados → consumidos → reconocidos, completando el ciclo de vida completo del procesamiento de eventos.
En conclusión, organizamos una canalización de enrutamiento de tareas dinámica y distribuida que procesa eventos en tiempo real con claridad y precisión. Fuimos testigos de cómo Kombu abstrae la complejidad de los sistemas de mensajería y al mismo tiempo nos brinda un control detallado sobre el enrutamiento, el consumo y la concurrencia de los trabajadores. A medida que vemos que los mensajes pasan del productor al intercambio, de la cola al trabajador, apreciamos más profundamente la elegancia del diseño de sistemas basados en eventos y ahora estamos bien equipados para escalar esta base a microservicios robustos, procesadores en segundo plano y flujos de trabajo de nivel empresarial.
Consulte los CÓDIGOS COMPLETOS. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín.
Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.