Uso del cluster in un modulo Node

Question

UPDATE: Anche se questo particolare scenario non è realistico, come per i commenti, sono ancora interessato a come si potrebbe scrivere un modulo che faccia uso del clustering senza ripetere il processo padre ogni volta .

---

Sto cercando di scrivere un modulo chiamato Node.js mass-request che accelera il gran numero di richieste HTTP, distribuendole ai processi figli.

mia speranza è che, al di fuori, funziona in questo modo.

var mr = require("mass-request"), 
    scraper = mr(); 

for (var i = 0; i < my_urls_to_visit.length; i += 1) { 
    scraper.add(my_urls_to_visit[i], function(resp) { 
     // do something with response 
    } 
} 

Per iniziare, ho messo insieme uno scheletro per il modulo di massa richiesta.

var cluster = require("cluster"), 
    numCPUs = require("os").cpus().length; 

module.exports = function() { 
    console.log("hello from mass-request!"); 
    if (cluster.isMaster) { 
     for (var i = 0; i < numCPUs; i += 1) { 
      var worker = cluster.fork();    
     } 

     return { 
      add: function(url, cb) {}  
     }  
    } else { 
     console.log("worker " + process.pid + " is born!"); 
    } 
} 

Poi ho provarlo in questo modo in uno script di test:

var m = mr(); 
console.log("hello from test.js!", m); 

mi aspettavo di vedere "Ciao da mass-richiesta!" registrato quattro volte (come infatti è). Con mio grande stupore, vedo anche "ciao da test.js" quattro volte. Chiaramente non capisco come funziona cluster.fork(). È rieseguire l'intero processo, non solo la funzione che lo chiama la prima volta?

Se sì, come si fa a fare uso di clustering in un modulo senza turbare la persona che utilizza quel modulo con una logica multi-processo disordinato?

Answer 1

credo che quello che stai cercando è in setupMaster

Dalla documentazione:

cluster.setupMaster ([Impostazioni])

• impostazioni Oggetto
  • exec Il percorso del file della stringa nel file worker. (Predefinito = process.argv [1])
  • args Argomenti di stringa di array passati a worker. (Predefinito = process.argv.slice (2))
  • silenzioso Booleano se inviare o meno l'output allo stdio del genitore. (Default = false)

setupMaster viene utilizzato per modificare il comportamento predefinito 'forchetta'. Una volta chiamato, le impostazioni saranno presenti in cluster.settings

Facendo uso della proprietà exec è possibile far avviare i lavoratori da un modulo diverso.

Importante: come lo stato dei documenti, questo può essere chiamato solo una volta. Se si sta dipendendo da questo comportamento per il modulo, il chiamante non può utilizzare cluster o l'intera cosa va in pezzi.

Ad esempio:

index.js

var cluster = require("cluster"), 
    path = require("path"), 
    numCPUs = require("os").cpus().length; 

console.log("hello from mass-request!"); 
if (cluster.isMaster) { 
    cluster.setupMaster({ 
    exec: path.join(__dirname, 'worker.js') 
    }); 

    for (var i = 0; i < numCPUs; i += 1) { 
    var worker = cluster.fork(); 
    } 

    return { 
    add: function (url, cb) { 
    } 
    } 
} else { 
    console.log("worker " + process.pid + " is born!"); 
} 

lavoratore.js

console.log("worker " + process.pid + " is born!"); 

uscita

node index.js 
hello from mass-request! 
worker 38821 is born! 
worker 38820 is born! 
worker 38822 is born! 
worker 38819 is born! 

Answer 2

Mentre è vero che la natura asincrona di node.js rende impressionante, corre ancora in un singolo thread sul server in un unico ciclo di eventi. Il multithreading di un'app node.js con cluster consente di inoltrare i processi secondari dell'app nei propri thread, consentendoti di utilizzare meglio un server multi-core. Avevo costruito un'architettura di server di gioco qualche tempo fa che utilizzava cluster e zmq (ZeroMQ) per il multithread e permetteva ai processi di inviare facilmente messaggi avanti e indietro su vari canali. Ho semplificato questa architettura nell'esempio seguente per sperare che aiuti a illustrare come il file node.js multithreading può essere messo insieme. Mi scuso se è un po 'approssimativo, era anni fa ed ero relativamente nuovo al nodo in quel momento;)

Idealmente, non si desidera nidificare tutto per il master/figlio in un singolo script, ma io immaginato che questo fosse il modo più semplice per permetterti di copiare/incollare/eseguire :)

Come hai detto nel tuo commento, ho dato un buon esempio di clustering, ma non uno che si adatta al tuo caso d'uso specifico per quanto riguarda la spedizione di tutto ciò che circonda . Non ho avuto molto tempo, quindi ho adattato il mio esempio per farlo funzionare in modo abbastanza rapido. Dare a questo un colpo:

mass-request.js

var cluster = require('cluster'); 
var zmq = require('zmq'); 

module.exports = { 
    _childId : null, 
    _urls : [], 
    _threadCount : 1, 
    _readyThreads : 0, 
    _callbacks : {}, 
    zmqReceive : null, //the socket we receive on for this thread 
    zmqMaster : null, //the socket to the master 
    zmqChildren : {}, //an object storing the sockets for the children 
    setThreads : function(threadCount) { 
     this._threadCount = threadCount; 
    }, 
    add : function(url , cb) { 
     this._urls.push({url: url, cb : cb }); 
    }, 
    run : function() { 

     if(cluster.isMaster) { 

      this._masterThread(); 

     } else { 

      this._childThread(); 

     } 

    }, 
    _masterThread : function() { 

     console.log('Master Process Starting Up'); 

     this.zmqReceive = zmq.socket('pull').bindSync('ipc://master.ipc'); 

     //bind handler for messages coming into this process using closure to allow us to access the massrequest object inside the callback 
     (function(massRequest) { 
      this.zmqReceive.on('message' , function(msg) { 

       msg = JSON.parse(msg); 

       //was this an online notification? 
       if(msg && msg.status == 'Online') { 
        massRequest._threadReady(); 
        return; //we're done 
       } 
       if(msg && msg.html) { 
        //this was a response from a child, call the callback for it 
        massRequest._callbacks[ msg.sender ].call(massRequest , msg.html); 
        //send the child another URL 
        massRequest._sendUrlToChild(msg.sender); 
       } 

      }); 
     }).call(this , this); 

     //fork 4 child processes and set up the sending sockets for them 
     for(var i=0; i < this._threadCount; ++i) { 
      //set up the sending socket 
      this.zmqChildren[i] = zmq.socket('push').connect('ipc://child_' + i + '.ipc'); 
      //fork the process and pass it an id 
      cluster.fork({ 
       _childId:i 
      }); 
     } 

    }, 
    _sendUrlToChild : function(child) { 
     //if there's no urls left, return (this would also be a good place to send a message to the child to exit gracefully) 
     if(!this._urls.length) return; 
     //grab a url to process 
     var item = this._urls.pop(); 
     //set the callback for the child 
     this._callbacks[child] = item.cb; 
     this.zmqChildren[child].send(JSON.stringify({ url:item.url })); 
    }, 
    _processUrls : function() { 
     for(var i=0; i < this._threadCount; ++i) { 
      this._sendUrlToChild(i); 
     } 
    }, 
    _threadReady : function() { 
     if(++this._readyThreads >= this._threadCount) { 
      //all threads are ready, send out urls to start the mayhem 
      console.log('All threads online, starting URL processing'); 
      this._processUrls(); 
     } 
    }, 
    _childProcessUrl : function(url) { 
     console.log('Child Process ' + this.childId + ' Handling URL: ' + url); 
     //do something here to scrape your content however you see fit 
     var html = 'HTML'; 
     this.zmqMaster.send(JSON.stringify({ sender:this.childId, html:html })); 
    }, 
    _childThread : function() { 

     //get the child id that was passed from cluster 
     this.childId = process.env._childId; 

     console.log('Child Process ' + this.childId + ' Starting Up'); 

     //bind the pull socket to receive messages to this process 
     this.zmqReceive = zmq.socket('pull').bindSync('ipc://child_' + this.childId + '.ipc'); 

     //bind the push socket to send to the master 
     this.zmqMaster = zmq.socket('push').connect('ipc://master.ipc'); 

     //bind handler for messages coming into this process 
     (function(massRequest) { 
      this.zmqReceive.on('message' , function(msg) { 

       msg = JSON.parse(msg); 

       console.log('Child ' + this.childId + ': ' + msg); 

       //handle the url 
       if(msg && msg.url) massRequest._childProcessUrl(msg.url); 

      }); 
     }).call(this , this); 

     //let the master know we're done setting up 
     this.zmqMaster.send(JSON.stringify({sender:this.childId,status:'Online'})); 

    }, 
} 

demo.js

var mr = require('./mass-request.js'); 
mr.setThreads(4); 
mr.add('http://foo.com' , function(resp) { 
    console.log('http://foo.com is done'); 
}); 
mr.add('http://bar.com' , function(resp) { 
    console.log('http://bar.com is done'); 
}); 
mr.add('http://alpha.com' , function(resp) { 
    console.log('http://alpha.com is done'); 
}); 
mr.add('http://beta.com' , function(resp) { 
    console.log('http://beta.com is done'); 
}); 
mr.add('http://theta.com' , function(resp) { 
    console.log('http://theta.com is done'); 
}); 
mr.add('http://apples.com' , function(resp) { 
    console.log('http://apples.com is done'); 
}); 
mr.add('http://oranges.com' , function(resp) { 
    console.log('http://oranges.com is done'); 
}); 
mr.run(); 

mettere quelle nella stessa cartella ed eseguire node demo.js.

Vorrei anche sottolineare che, poiché la base di questo è stato tirato da uno dei miei altri progetti che utilizzavano [0MQ] [http://zeromq.org/], è necessario che installato accanto al [Node.JS modulo per esso] [https://github.com/JustinTulloss/zeromq.node]npm install zmq e ovviamente il modulo cluster.È possibile scambiare le parti ZMQ per qualsiasi altro metodo di comunicazione interprocesso che si desidera, naturalmente. Questo capita di essere uno che avevo familiarità con e aveva usato.

Breve panoramica: Il thread principale AKA lo script che chiama il metodo run() farà ruotare i bambini X (può essere impostato chiamando setThreads). Questi bambini riportano al thread principale tramite zeri ZeroMQ al termine dell'inizializzazione. Una volta che tutti i thread sono pronti, lo script master invia gli URL ai bambini in modo che possano eseguire il recupero e recuperare l'HTML. Restituiscono l'HTML al master dove lo trasferiscono nella funzione di callback appropriata per quell'URL e quindi invia un altro URL allo script figlio. Anche se non è una soluzione perfetta, le funzioni di callback stanno ancora andando a collo di bottiglia nel thread principale (master) perché non è possibile spostarle facilmente su un altro thread. Questi callback possono contenere chiusure/variabili/ecc che potrebbero non funzionare correttamente al di fuori del thread principale senza alcun tipo di meccanismo di condivisione degli oggetti.

Chiunque, se giri la mia piccola demo, vedrai 4 thread che "elaborano" gli URL (in realtà non caricano gli URL per semplicità).

Speriamo che aiuta;)