Interoperando con Django/Celery de Java

Question

Nuestra empresa tiene un sitio web basado en Python y algunos nodos de trabajo basados ​​en Python que se comunican a través de Django/Celery y RabbitMQ. Tengo una aplicación basada en Java que necesita enviar tareas a los trabajadores de Apio. Puedo enviar trabajos a RabbitMQ desde Java sin problemas, pero los trabajadores con base en Apio nunca recogen los trabajos. Al observar las capturas de paquetes de ambos tipos de envíos de trabajos, existen diferencias, pero no puedo comprender cómo explicarlas porque una gran parte es binaria y no puedo encontrar documentación sobre la decodificación. ¿Alguien aquí tiene alguna referencia o experiencia al tener Java/RabbitMQ y Celery trabajando juntos?

Answer 1

Encontré la solución. La biblioteca de Java para RabbitMQ se refiere a intercambios/colas/routekeys. En Celery, el nombre de la cola en realidad está mapeando al intercambio al que se hace referencia en la biblioteca de Java. Por defecto, la cola para el apio es simplemente "apio". Si la configuración de Django definen una cola llamada "myQueue" utilizando la siguiente sintaxis:

CELERY_ROUTES = { 
    'mypackage.myclass.runworker'  : {'queue':'myqueue'}, 
} 

continuación, el código basado en Java tiene que hacer algo como lo siguiente:

 ConnectionFactory factory = new ConnectionFactory(); 
     Connection connection = null ; 
     try { 
      connection = factory.newConnection(mqHost, mqPort); 
     } catch (IOException ioe) { 
      log.error("Unable to create new MQ connection from factory.", ioe) ; 
     } 

     Channel channel = null ; 
     try { 
      channel = connection.createChannel(); 
     } catch (IOException ioe) { 
      log.error("Unable to create new channel for MQ connection.", ioe) ; 
     } 

     try { 
      channel.queueDeclare("celery", false, false, false, true, null); 
     } catch (IOException ioe) { 
      log.error("Unable to declare queue for MQ channel.", ioe) ; 
     } 

     try { 
      channel.exchangeDeclare("myqueue", "direct") ; 
     } catch (IOException ioe) { 
      log.error("Unable to declare exchange for MQ channel.", ioe) ; 
     } 

     try { 
      channel.queueBind("celery", "myqueue", "myqueue") ; 
     } catch (IOException ioe) { 
      log.error("Unable to bind queue for channel.", ioe) ; 
     } 

      // Generate the message body as a string here. 

     try { 
      channel.basicPublish(mqExchange, mqRouteKey, 
       new AMQP.BasicProperties("application/json", "ASCII", null, null, null, null, null, null, null, null, null, "guest", null, null), 
       messageBody.getBytes("ASCII")); 
     } catch (IOException ioe) { 
      log.error("IOException encountered while trying to publish task via MQ.", ioe) ; 
     } 

Resulta que es justo una diferencia en la terminología.