使用 AMQP 在 Node.js 和 Django 之间传输对象答案

【问题标题】：Object transfer between Node.js and Django using AMQP使用 AMQP 在 Node.js 和 Django 之间传输对象
【发布时间】：2012-04-24 19:26:50
【问题描述】：

我从 websockets 开始，我对数字库和配置选项感到非常困惑。我只想设置一个项目，其中 node.js 服务器调用 python/django 中的方法，当最后一个完成时，它将结果传输回 node.js 服务器。这是我到目前为止所拥有的：

来自this tutorial的Nodes.js AMQP：

var conn = amqp.createConnection();
conn.on('ready', function(){
 var exchange = conn.exchange('?1', {'type': 'fanout', durable: false}, function() {
     exchange.publish('?2', {add: [1,2]});
 });
});

来自this tutorial的Django Celery：

from celery.decorators import task

@task()
def add(x, y):
   return x + y

我不知道这是否是要走的路，如果有人能阐明这个问题，我会很高兴。

--- 编辑

我成功地使用 AMQP 进行了简单的字符串传输：

test.py

 import pika
 connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost'))
 channel = connection.channel()

 channel.queue_declare(queue='task_queue', durable=True)

 print ' [*] Waiting for messages. To exit press CTRL+C'

 def callback(ch, method, props, body):
     print " [x] Received %r" % (body,)
     response = body + " MODIFIED"
     #response = get_a_concept()
     print " [x] Done"
     ch.basic_publish(exchange='',
                 routing_key=props.reply_to,
                 properties=pika.BasicProperties(correlation_id = \
                                                 props.correlation_id),
                 body=str(response))
     ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                  queue='task_queue')

channel.start_consuming()

app.js

 var connection = amqp.createConnection({ host: 'localhost' });
 connection.addListener('ready', function() {


var exchange = connection.exchange('', {
    'type' : 'direct',
    durable : false
}, function() {

    var queue = connection.queue('incoming', {
        durable : false,
        exclusive : true }, function() {
        queue.subscribe(function(msg) {
            console.log("received message: ");
            console.log(msg.data.toString());
        });

    });

    exchange.publish('task_queue', "it works!", {
        'replyTo' : 'incoming'
    });
});

});

不过，我不确定这是否是最好的实现，我什至没有在这里使用 queue.bind() 方法。当我尝试传递一个复杂的对象（json 甚至是一个简单的数组）时，就会出现问题。改变这一行

 body= (["a","b","c"])#str(response))

导致以下错误：

 Traceback (most recent call last):
   File "test.py", line 56, in <module>
     channel.start_consuming()
   File "/Library/Python/2.7/site-packages/pika/adapters/blocking_connection.py", line 293, in      start_consuming
 (...)
   File "/Library/Python/2.7/site-packages/pika/simplebuffer.py", line 62, in write
     self.buf.write(data)
 TypeError: must be string or read-only character buffer, not list

有没有序列化复杂对象的解决方案？我错过了什么吗？

【问题讨论】：

我不熟悉节点，但它应该有某种异步 http 库。从你对你想要的东西的描述看来，芹菜可能是不必要的。如果您不需要 celery 提供的功能，只需创建一个 django 视图，将您需要的数据返回给 node.js 并从节点调用其 url！
谢谢。可以按照您的建议进行http resquests，我会尝试的。但就我而言，我有 django 的长期任务（不是这个添加），恐怕当很多客户端发出大量长期任务请求时，http 请求不适合..
使用json之类的东西来序列化数据怎么样？查看 python 的 simplejson（虽然不确定 node.js 方面）。

标签： django node.js amqp django-celery pika

【解决方案1】：

body 必须是字节数组！

你必须二进制序列化你的对象。但是 nods.jss 和 pythons 可能不兼容。

蟒蛇：pickle node.js: rucksack on of...

一种快速的解决方法是将您的数组加入到这样的字符串中：

test.py

...
body = ';'.join(["a","b","c"]) # which will result in this string 'a;b;c'
...

并在您的 app.js 中将该字符串拆分为一个数组。

...
result = stringResult.split(';'); //stringResult is 'a;b;c'
...

另一种选择可能是在 node.js 中使用 'async' 执行该异步任务

【讨论】：