Tengo un módulo en Node.js que selecciona repetidamente un documento de MongoDB y lo procesa. Un documento debe procesarse una sola vez. También quiero usar el concepto de procesos múltiples. Quiero ejecutar el mismo módulo (proceso) en diferentes procesadores, que se ejecutan de forma independiente.
El problema es que puede haber un escenario en el que dos trabajadores diferentes recojan y procesen el mismo documento. Cómo varios procesos pueden saber que un documento en particular es procesado por otro trabajador, por lo que no debo tocarlo. Y no hay forma de que mis procesos independientes puedan comunicarse. No puedo usar un padre que bifurca múltiples procesos y actúa como un puente entre ellos. ¿Cómo evitar este tipo de problemas en Node.js?
Una forma de hacerlo es asignar una identificación numérica única a cada uno de sus documentos MongoDB y asignar un identificador numérico único a cada uno de sus trabajadores de node.js.
Por ejemplo, tenga un env var llamado NUM_WORKERS, y luego en su módulo node.js:
var NumWorkers = process.env.NUM_WORKERS || 1;Luego debe asignar una identificación de número de instancia única y contigua (en el rango de 0 a NumWorkers-1) a cada uno de sus trabajadores (por ejemplo, a través de un parámetro de línea de comando leído por su proceso node.js cuando se inicializa). Puede almacenar eso en una variable llamada MyWorkerInstanceNum.
Cuando elija un documento de MongoDB, llame a la siguiente función (pasando el ID de documento único del documento como parámetro):
function isMine(documentId){ // // Example: documentId=10 // NumWorkers= 4 // (10 % 4) = 2 // If MyWorkerInstanceNum is 2, return true, else return false. return ((documentId % NumWorkers) === MyWorkerInstanceNum); }Solo continúe procesando el documento si isMine() devuelve verdadero. Por lo tanto, varios trabajadores pueden "elegir" un documento, pero solo un trabajador lo procesará.
Simplemente mantenga un registro de transacciones del documento que se está procesando por su identificación única. En la tabla de registro de transacciones para los documentos procesados, escriba el estado como uno de los siguientes (por ejemplo):
requested initiated processed failedTambién puede querer una columna en esa tabla para stderr/stdout en caso de que quiera saber por qué algo falló o tuvo éxito, y las marcas de tiempo, ese tipo de cosas.
Cuando inicie el procesamiento del documento en su aplicación Node, busque el documento por ID y verifique su estado. Si no existe, entonces eres libre de procesarlo.
Pseudocódigo (lo siento, ¡no soy un tipo de Mongo!):
db.collection.list('collectionName', function(err, doc) { db.collection.find(doc.id, 'transactions', function(err, trx) { if (trx === undefined || trx.status === 'failed') { DocProcessor.child.process(doc) } else { // don't need to process it, it's already been done } }) })También querrá habilitar el bloqueo de simultaneidad en la recopilación de registros de transacciones para asegurarse de que una fila (y el trabajo posterior) no se pueda duplicar. Si esto se convierte en un desafío para garantizar que los documentos se coloquen en la cola correctamente, considere agregar un servicio AMQP para manejar la cola de los documentos. Configure un controlador para administrar la distribución de los procesos secundarios y el registro de transacciones. El flujo sería algo como:
MQ ⇢ Log ⇢ Handler ⇢ Hijos del procesador de documentos