Tương tác với Django/Celery Từ Java

Question

Công ty chúng tôi có một trang web dựa trên Python và một số nút công nhân dựa trên Python giao tiếp qua Django/Celery và RabbitMQ. Tôi có một ứng dụng dựa trên Java mà cần phải gửi nhiệm vụ cho các công nhân dựa trên Celery. Tôi có thể gửi công việc đến RabbitMQ từ Java, nhưng công nhân dựa trên Celery không bao giờ chọn công việc. Từ việc xem xét các gói chụp của cả hai loại công việc gửi, có sự khác biệt, nhưng tôi không thể hiểu được làm thế nào để giải thích cho chúng bởi vì rất nhiều nó là nhị phân mà tôi không thể tìm thấy tài liệu về giải mã. Có ai ở đây có bất kỳ tài liệu tham khảo hoặc kinh nghiệm với việc có Java/RabbitMQ và Celery làm việc cùng nhau?

Answer 1

Tôi đã tìm ra giải pháp. Thư viện Java cho RabbitMQ dùng để chỉ trao đổi/hàng đợi/routekeys. Trong Celery, tên hàng đợi thực sự là ánh xạ tới trao đổi được đề cập trong thư viện Java. Theo mặc định, hàng đợi cho Celery đơn giản là "cần tây". Nếu cài đặt Django của bạn xác định một hàng đợi gọi là "myqueue" bằng cách sử dụng cú pháp sau:

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

Sau đó, các mã Java dựa cần phải làm điều gì đó như sau:

 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) ; 
     } 

Nó chỉ ra rằng nó chỉ là sự khác biệt về thuật ngữ.