java - 从 Java 与 Django/Celery 互操作

Question

我们公司有一个基于 Python 的网站和一些基于 Python 的工作节点，它们通过 Django/Celery 和 RabbitMQ 进行通信。我有一个基于 Java 的应用程序，需要将任务提交给基于 Celery 的工作人员。我可以很好地将工作从 Java 发送到 RabbitMQ，但是基于 Celery 的工作人员永远不会接手这些工作。通过查看两种类型的作业提交的数据包捕获，存在差异，但我无法理解如何解释它们，因为很多都是二进制的，我找不到有关解码的文档。这里有人对 Java/RabbitMQ 和 Celery 协同工作有任何引用或经验吗？

Answer 1

我找到了解决方案。 RabbitMQ 的 Java 库指的是交换器/队列/路由键。在 Celery 中，队列名称实际上映射到 Java 库中引用的交换器。默认情况下，Celery 的队列就是“celery”。如果您的 Django 设置使用以下语法定义了一个名为“myqueue”的队列:

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

然后基于 Java 的代码需要执行如下操作:

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

事实证明，这只是术语上的差异。