使用 Node.js 接收 AWS SQS 消息并根据收到的消息调用函数

Question

我正在使用 Node.js 服务器与带有 Amazon SQS 的 python DAO 对话。我能够向 DAO 发送 SQS，但不知道如何在 python DAO 上发回一些东西并在 Node.js 服务器上正确监听它。我也很好奇如何根据从 python DAO 收到的 Node.js SQS 消息来调用另一个函数。任何帮助将不胜感激！到目前为止，这是我的代码：

Node.js 监听 SQS 代码：

app.get('/readDAOSQSMessage', function(req, res) {
  var params = {
   QueueUrl: DAO_QUEUE_URL,
   MaxNumberOfMessages: 1, // how many messages do we wanna retrieve?
   VisibilityTimeout: 60, // seconds - how long we want a lock on this job
   WaitTimeSeconds: 3 // seconds - how long should we wait for a message?
  };
sqs.receiveMessage(params, function(err, data) {
   if (data.Messages) {
      var message = data.Messages[0],
      body = JSON.parse(message.Body);
      res.send("Username: " + body.username + " Password: " + body.password);
      removeFromQueue(message, DAO_QUEUE_URL); 
   }
 });
});

Python 发送 SQS 代码： queue = conn.get_queue('DAO-Queue')

    writeQueue = conn.get_queue('DaoToServerQueue')
    message = RawMessage()
    message.set_body("This is my first message.")
    writeQueue.write(message)

    while True:
    rs = queue.get_messages()
    if len(rs) > 0:
        m = rs[0]
        print("Message Body: " + m.get_body())
        message = m.get_body()
            #unicodedata.normalize('NFKD',message).encode('ascii','ignore')
        #command = 'dao.'+message[1:-1]
            message = message.encode('ascii','ignore')
            print "message: "+message[1:-1]
            print "backslash: "+'\\'
        replace_str = '\\'+'"'
            print replace_str
            print message.find(replace_str)
            print"\nmessage type: "+str(type(message))
            message = message.replace(replace_str,'"')
            print "\nnew message: "+message
            command = 'dao.'+message[1:-1]
            print "\n command: "+command
        #unicodedata.normalize('NFDK',message).encode('ascii','ignore')
        #command = 'dao.'+ message
            #unicodedata.normalize('NFDK',command).encode('ascii','ignore')
            eval(command)
        print("Command: " + command)
        queue.delete_message(m)

Answer 1

通常您希望通过长轮询来收听接收方的消息。

function readMessages() {
    async.parallel([
        pollMessage,
    ], function(err, callbacks) {
        readMessages();
    });
}

function pollMessage(callback) {
    sqs.receiveMessage({
        QueueUrl: receiveQueue,
        WaitTimeSeconds: 5,
        MaxNumberOfMessages: 1,
        // VisibilityTimeout: 20
    }, function(err, data) {
        if (err) console.log(err);
        if (!data) {
            callback();
            return;
        }
        if (data.Messages) {
            // Get the first message (should be the only one since we said to only get one above)
            for (var i = 0; i < data.Messages.length; i++) {
                var message = data.Messages[i];
                removeFromQueue(message, callback);
                // Now this is where you'd do something with this message
                handleIncomeMessage(message); // whatever you wanna do
                // Clean up after yourself... delete this message from the queue, so it's not executed again
            }
        } else {
            callback();
        }
    });
}

要在 NodeJS 服务器上作为 lambda 函数执行的代码。 您可以通过将消息类型附加到有效负载来决定如何处理消息。注意上面的 handleIncomeMessage() 函数，它可以是这样的：

function handleIncomeMessage(data) {
    var msg = JSON.parse(data.Body);
    var type = msg.type;
    console.log("Got a new sqs message: ", msg.type);
    switch (type) {
        case 'someCoolType':
            {
                Manager.proceedWithCoolType(msg.restOfPayload);
                break;
            }
    }
}