Nodejs file d'attente sqs processeur

Je suis en train d'écrire un nodejs file d'attente sqs processeur.

"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();

Le code ci-dessus fonctionne très bien pour un seul message dans la file d'attente. Comment dois-je écrire ce script pour qu'il continue d'interrogation pour les messages en file d'attente jusqu'à ce que tous les messages sont traités? Dois-je utiliser set timeout?

OriginalL'auteur Yalamber | 2013-07-18

amazon-sqs node.js

15

Tout d'abord, vous devriez définitivement utilisation le long du scrutin technique fournie par Amazon, et comme je comprends que vous êtes déjà à l'utiliser parce que vous avez "WaitTimeSeconds": 20 argument en sqs.receiveMessage appel. J'espère que vous n'oubliez pas de le configurer dans le AWS interface Web.

Au sujet d'interrogation pour les messages, vous pouvez utiliser différentes techniques, dont la minuteries, mais je pense que le plus simple sera tout simplement appeler votre readMessage() fonction à la fin de receiveMessage's (ou même execs') la fonction callback. Afin de traitement (ou d'attente) le message suivant dans la file d'attente va commencer immédiatement après la fin du traitement précédent message dans la file d'attente.

Mise à JOUR:

Que pour moi dans votre nouvelle version de code il y a de nombreux readMessage() appels. Je pense que c'est mieux de le réduire à garder le code plus clair et facile à entretenir. Mais si vous partez, par exemple, le seul à appeler à la fin de votre principal receiveMessage callback, vous recevrez un lot de PHP travailleur scripts s'exécutant en parallèle - et c'est peut-être pas si mauvaise du point de vue de la performance, mais vous devrez ajouter un peu compliqué script pour contrôler la quantité de parallèle travailleurs. Je pense que vous pouvez couper quelques appels de exec rappel, essayez de joindre ifs et joindre les appels en principal de rappel.
```
"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,
"MaxNumberOfMessages": 1,
"VisibilityTimeout": 30,
"WaitTimeSeconds": 20
}, function (err, 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+'"',
function (error, stdout, 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();
```
Propos des fuites de mémoire: je pense que vous ne devriez pas vous inquiéter, car le prochain appel de readMessage() qui se passe dans la fonction de rappel donc pas de manière récursive, et appelée récursivement la fonction retourne la valeur à la fonction parent, juste après l'appel de receiveMessage() fonction.

Bonjour, merci de consulter mon ce gist. gist.github.com/yalamber/374add88e887e688d818
Aussi dois-je craindre une fuite de mémoire lors de l'exécution de ce script?
Mis à jour!
Vieux post, mais avez-vous rencontré des problèmes avec la fuite de mémoire, parce que je suis en utilisant presque la même idée avec Bluebird promesses. Et sur chaque sondage je peux voir que le segment est en croissance d'environ 200ko.
Je pense que cette approche encore des fuites de mémoire.

OriginalL'auteur zavg

Si vous utilisez le nœud, l'utilisation https://www.npmjs.com/package/sqs-worker module. Il fera le travail pour vous.

var SQSWorker = require('sqs-worker')
var options =
{ url: 'https://sqs.eu-west-1.amazonaws.com/001123456789/my-queue'
}
var queue = new SQSWorker(options, worker)
function worker(notifi, done) {
var message;
try {
message = JSON.parse(notifi.Data)
} catch (err) {
throw err
}
//Do something with `message` 
var success = true
//Call `done` when you are done processing a message. 
//If everything went successfully and you don't want to see it any more, 
//set the second parameter to `true`. 
done(null, success)
}

pourquoi le vote? github.com/BBC/sqs-consumer est une alternative aussi bien

OriginalL'auteur Madura Pradeep

Vous devez vous connecter pour publier un commentaire.