node.js – Nodejs sqs队列处理器

"use strict";
var appConf = require('./config/appConf');
var AWS = require('aws-sdk');
AWS.config.loadFromPath('./config/aws_config.json');
var sqs = new AWS.SQS();
var exec = require('child_process').exec;
function readMessage() {
  sqs.receiveMessage({
    "QueueUrl": appConf.sqs_distribution_url,"MaxNumberOfMessages": 1,"VisibilityTimeout": 30,"WaitTimeSeconds": 20
  },function (err,data) {
    var sqs_message_body;
    if (data.Messages) {
      if (typeof data.Messages[0] !== 'undefined' && typeof data.Messages[0].Body !== 'undefined') {
        //sqs msg body
        sqs_message_body = JSON.parse(data.Messages[0].Body);
        //make call to nodejs handler in codeigniter
        exec('PHP '+ appConf.CI_FC_PATH +'/index.PHP nodejs_handler make_contentq_call "'+ sqs_message_body.contentq_cat_id+'" "'+sqs_message_body.cnhq_cat_id+'" "'+sqs_message_body.network_id+'"',function (error,stdout,stderr) {
            if (error) {
              throw error;
            }
            console.log('stdout: ' + stdout);
            if(stdout == 'Success'){
              //delete message from queue
              sqs.deleteMessage({
                "QueueUrl" : appConf.sqs_distribution_url,"ReceiptHandle" :data.Messages[0].ReceiptHandle
              });
            }
          });
      }
    }
  });
}
readMessage();

首先,您应该使用Amazon提供的长轮询技术,并且据了解,您已经使用它,因为您有“WaitTimeSeconds”：在sqs.receiveMessage调用中有20个参数.我希望你没有忘记在 AWS Web interface配置它.

关于轮询消息 – 您可能会使用不同的技术,包括定时器,但我认为最简单的只是在receiveMessage(或甚至exec)回调函数的末尾调用readMessage()函数.因此,处理(或等待)队列中的下一个消息将在队列中处理先前消息的结束后立即开始.

更新：

至于我在你的新版本的代码中有很多readMessage()调用.我认为最好把它保持在更加清晰和易于维护的状态.但是,如果您离开,例如,在主要的receiveMessage回调结束时只有一个调用,您将收到许多并行运行的PHP工作脚本 – 也许从性能的角度来看,它不是那么糟糕 – 但是将不得不添加一些复杂的脚本来控制并行工作人员的数量.我想你可以在exec回调中调用一些调用,尝试加入ifs并在主回调中加入调用.

"use strict";
var appConf = require('./config/appConf');
var AWS = require('aws-sdk');
AWS.config.loadFromPath('./config/aws_config.json');
var delay = 20 * 1000;
var sqs = new AWS.SQS();
var exec = require('child_process').exec;
function readMessage() {
  sqs.receiveMessage({
    "QueueUrl": appConf.sqs_distribution_url,data) {
    var sqs_message_body;
    if (data.Messages) 
      && (typeof data.Messages[0] !== 'undefined' && typeof data.Messages[0].Body !== 'undefined')) {
        //sqs msg body
        sqs_message_body = JSON.parse(data.Messages[0].Body);
        //make call to nodejs handler in codeigniter
        exec('PHP '+ appConf.CI_FC_PATH +'/index.PHP nodejs_handler make_contentq_call "'+ sqs_message_body.contentq_cat_id+'" "'+sqs_message_body.cnhq_cat_id+'" "'+sqs_message_body.network_id+'"',stderr) {
            if (error) {
              // error handling 
            }
            if(stdout == 'Success'){
              //delete message from queue
              sqs.deleteMessage({
                "QueueUrl" : appConf.sqs_distribution_url,"ReceiptHandle" :data.Messages[0].ReceiptHandle
              },function(err,data){                
              });
            }
            readMessage();                
          });
      }          
    }        
    readMessage();        
  });
}
readMessage();

关于内存泄漏：我认为你不应该担心,因为下一次调用readMessage()发生在回调函数中 – 所以不是递归的,递归调用函数只是在调用receiveMessage()函数之后返回父函数的值.