Le discussioni sono appesi una sfida comune nello sviluppo di software che è per l'interfacciamento con dispositivi proprietari mediante interfacce proprietarie o standard come SNMP, Q3, o Telnet. Questo problema non si limita alla gestione della rete, ma si verifica in una vasta gamma di settori come server web, i processi invocando chiamate di procedura remota, e così via.
Un thread che avvia una richiesta a un dispositivo necessita di un meccanismo per rilevare nel caso in cui il dispositivo non risponde o risponde solo in parte. In alcuni casi in cui viene rilevato come un blocco, una specifica azione deve essere presa. L'azione specifica potrebbe essere o un nuovo processo o lasciando l'utente finale a conoscenza del fallimento compito o qualche opzione di recupero. In alcuni casi in cui un gran numero di compiti deve essere licenziato per un gran numero di elementi di rete da un componente, appeso rilevamento filo è importante in modo che non diventi un collo di bottiglia per l'elaborazione altro compito. Quindi ci sono due aspetti di gestione delle discussioni appesi: le prestazioni e la notifica .
Per l' aspetto notifica si può adattare il modello Observer Java per adattarsi al mondo multithread.
Sartoria del pattern Observer Java per sistemi multithread
A causa di sospensione attività, utilizzando le Java ThreadPoolclasse con una strategia di adattamento è la prima soluzione che viene in mente. Tuttavia con il Java ThreadPool nel contesto di alcuni fili appesi a caso per un periodo di tempo dà comportamenti indesiderati in base alla particolare strategia utilizzati, come la fame filo nel caso di strategia fissa pool di thread. Ciò è dovuto principalmente al fatto che il Java ThreadPool non ha un meccanismo per rilevare un blocco filo.
Potremmo provare un pool di cache filo, ma ha anche dei problemi.Se c'è un alto tasso di cottura compito, e alcuni appendere le discussioni, il numero di thread potrebbe sparare alto, provocando infine la fame di risorse e out-of-memory eccezioni. Oppure potremmo usare un Custom ThreadPool strategia di invocare una CallerRunsPolicy . In questo caso, un blocco filo potrebbe causare tutte le discussioni da appendere alla fine. (Il thread principale non dovrebbe mai essere il chiamante, poiché non vi è la possibilità che qualsiasi compito passato al thread principale può bloccarsi, causando tutto per fermerebbe.)
Quindi, qual è la soluzione? Io dimostrare un modello di non-così-semplice ThreadPool che regola la dimensione del pool in base al tasso di attività e in base al numero di fili appesi. Vediamo prima andare al problema di individuare le discussioni impiccagione.
Rilevare Discussioni Hanging
La figura 1 mostra una astrazione del modello:
Ci sono due classi importanti: ThreadManager e ManagedThread . Entrambi si estendono dal Java discussione classe. IlThreadManager detiene un contenitore che contiene il ManagedThreads . Quando un nuovo ManagedThread viene creato esso si aggiunge a questo contenitore.
ThreadHangTester testthread = new ThreadHangTester ("threadhangertest", 2000, false);
testthread.start ();
thrdManger.manage (testthread, ThreadManager.RESTART_THREAD, 10);
thrdManger.start ();Il ThreadManager itera su questa lista e chiama il ManagedThread s ' isHung () metodo. Questo è fondamentalmente una logica di controllo timestamp.
if (System.currentTimeMillis () - lastprocessingtime.get () maxprocessingtime>)
{
logger.debug ("La discussione è appeso");
return true;
}Se trova che un thread è andato in un ciclo di attività e mai aggiornato i suoi risultati ci vuole un meccanismo di recupero come previsto dal ManageThread .
while (IsRunning)
{
for (Iterator iteratore = managedThreads.iterator (); iterator.hasNext ();) {
ManagedThreadData thrddata = (ManagedThreadData) Iterator.next (); if (thrddata.getManagedThread () isHung (). ) { logger.warn ("Il filo di Hang rilevato per ThreadName =" + thrddata.getManagedThread () getName ());. switch (thrddata.getManagedAction ()) { RESTART_THREAD caso: / / L'azione è quello di riavviare il filo / / togliere dal direttore iterator.remove (); / / interrompere il trattamento di questa discussione, se possibile thrddata.getManagedThread () stopProcessing ();. if (. thrddata.getManagedThread () getClass () == ThreadHangTester.class) / / Per sapere quale tipo di filo per creare { ThreadHangTester newThread = new ThreadHangTester ("restarted_ThrdHangTest", 5000, true); / / Crea un nuovo thread newThread.start (); / / aggiunge di nuovo a essere gestito gestire (newThread, thrddata. getManagedAction (), thrddata.getThreadChecktime ()); } break; .........
Per una nuova ManagedThread essere creati e utilizzati al posto del appeso uno non dovrebbe contenere qualsiasi stato o qualsiasi contenitore. Per questo il contenitore su cui il ManagedThread atti dovrebbero essere separati. Qui stiamo usando il ENUM basate modello Singleton per contenere l'elenco attività. Così il recipiente contenente i compiti è indipendente dal thread di elaborazione dei compiti. Fare clic sul seguente link per scaricare i sorgenti per il modello descritto: Fonte discussione Java Manager .
Hanging Fili e Java Strategie ThreadPool
Il Java ThreadPool non ha un meccanismo per rilevare le discussioni impiccagione. Usando una strategia simile fissa ThreadPool ( Executors.newFixedThreadPool () ) non funzionano perché se alcune operazioni appendere nel tempo, tutti i thread sarà alla fine in uno stato bloccato. Un'altra opzione è utilizzando un criterio di cache ThreadPool (Executors.newCachedThreadPool () ). Ciò potrebbe garantire che ci saranno sempre thread disponibili per elaborare un compito, solo limitata dalla VM di memoria, CPU e limiti thread. Tuttavia, con questa politica non c'è controllo del numero di thread che vengono creati. Indipendentemente dal fatto che un thread di elaborazione si blocca e non, utilizzando questo criterio, mentre il tasso di attività è alto porta ad un numero enorme di thread ottenere creato. Se non si ha abbastanza risorse per la JVM molto presto avrete raggiunto la soglia massima di memoria o CPU. È abbastanza comune vedere il numero di thread colpito centinaia o migliaia. Anche se sono rilasciati una volta il compito è stato elaborato, a volte durante la raffica di gestione l'elevato numero di thread sarà sopraffare risorse di sistema.
Una terza opzione è utilizzando strategie personalizzate o politiche. Una delle ipotesi è quella di avere un pool di thread che scala da 0 a qualche numero massimo. Quindi, anche se un filo appeso un nuovo thread si verrebbe a creare fino a quando il numero di thread massima è stata raggiunta:
execexec = ThreadPoolExecutor nuovi (0, 3, 60, TimeUnit.SECONDS, nuove SynchronousQueue ()); Qui 3 è il filo numero massimo e keep-alive tempo è impostato a 60 secondi come questo è un compito processo ad alta intensità. Se diamo uno abbastanza alto numero di thread massimo, questo è più o meno una politica ragionevole da utilizzare nel contesto delle attività sospese. L'unico problema è che se i fili appesi non vengono rilasciati alla fine c'è una piccola possibilità che tutte le discussioni potrebbero ad un certo punto d'aggancio. Se le discussioni massimo è sufficientemente elevato e nell'ipotesi che un blocco del compito è un fenomeno raro, allora questa politica misura la fattura.
Sarebbe stato dolce se il ThreadPool aveva anche un meccanismo di innesto di rilevare le discussioni impiccagione. Parlerò un design così tardi. Naturalmente se tutti i fili sono congelato fino è possibile configurare e utilizzare il compito respinto-politica del pool di thread. Se non si vogliono scartare le attività che avrebbe dovuto usare il CallerRunsPolicy :
execexec = ThreadPoolExecutor nuovi (0, 20, 20, TimeUnit.MILLISECONDS, nuove SynchronousQueue () nuovo
ThreadPoolExecutor.CallerRunsPolicy ());In questo caso, se un thread appendere causato un compito da respingere, tale compito sarebbe stato dato al thread chiamante da trattare. C'è sempre la possibilità di tale compito anche appeso. In questo caso l'intero processo si bloccherebbe. Quindi è meglio non aggiungere una tale politica in questo contesto.
NotificationProcessor public class implementa Runnable { notificationOrginator privato NotificationOriginator finale; IsRunning boolean = true; finale privato ExecutorService execexec; AlarmNotificationProcessor (norginator NotificationOriginator) {/ / ctor / / execexec = Executors.newCachedThreadPool ();// Troppe discussioni / / execexec = Esecutori . newFixedThreadPool (2 );//, non appendere le attività di rilevazione execexec = ThreadPoolExecutor nuovi (0, 4, 250, TimeUnit.MILLISECONDS, nuove SynchronousQueue (), ThreadPoolExecutor.CallerRunsPolicy nuovo ()); } public void run () { while (IsRunning) { try { compito finale task = TaskQueue.INSTANCE.getTask (); thisTrap Runnable = new Runnable () { public void run () { + + alarmid; notificaionOrginator.notify (OctetString new (), / / Task elaborazione nbialarmnew.getOID (), nbialarmnew.createVariableBindingPayload ()); É ........}}; execexec.execute (thisTrap); }
Un ThreadPool personalizzati con Hang Detection
Una biblioteca pool di thread con la capacità di rilevazione compito appendere e la gestione sarebbe bello avere. Ho sviluppato uno e io lo dimostrano al di sotto. Questo è in realtà una porta da un C + + pool di thread che ho progettato ed utilizzato alcune indietro nel tempo (vedi riferimenti). Fondamentalmente, questa soluzione utilizza il modello di comando e la catena di responsabilità modello. Tuttavia, per implementare il modello di comando in Java senza l'ausilio di supporto delle funzioni oggetto è un po 'difficile. Per questo ho dovuto cambiare l'implementazione leggermente utilizzare la reflection Java.Si noti che il contesto in cui è stato progettato questo modello era il luogo dove un pool di thread doveva essere montato in / inserito senza modificare nessuna delle classi esistenti. (Credo che quella grande vantaggio di programmazione orientata agli oggetti è che ci dà un modo di progettare le classi in modo da fare un uso efficace del principio Aperto Chiuso. Questo è particolarmente vero per complessi codice legacy vecchio e potrebbe essere meno rilevante per sviluppo di nuovi prodotti.) Quindi ho usato una riflessione invece di utilizzare un interfaccia per implementare il modello di comando. Il resto del codice potrebbe essere portato senza grandi cambiamenti, come quasi tutte le primitive di sincronizzazione dei thread e la segnalazione sono disponibili in Java 1.5 in poi.
public class Comando {
private Object [] argParameter;
........
/ / Ctor per un metodo con due argomenti
di comando (T pObj, String methodName, timeout lungo, chiave String, int arg1, arg2 int) {
m_objptr = pObj;
m_methodName = mthodName;
m_timeout timeout =;
m_key key =;
argParameter = new Object [2];
argParameter [0] = arg1;
argParameter [1] = arg2; } / / Chiama il metodo dell'oggetto eseguire vuoto () { classe klass m_objptr.getClass = (); Class [] paramTypes = new Class [] {int.class, int.class}; try { methodName Metodo = klass.getMethod (m_methodName, paramTypes); / / System.out . println ("Trovato il metodo ->" + methodName); if (argParameter.length == 2) { methodName.invoke (m_objptr, (Object) argParameter [0], Object () argParameter [1]); }
Esempio di utilizzo di questo modello:
public class CTask
.. {
public int DoSomething (int a, int b) {...}
}Comando
Ora abbiamo due classi più importanti qui. Uno è il ThreadChain classe, che implementa la catena del modello di Responsabilità:
public class ThreadChain implementa Runnable { pubblico ThreadChain (ThreadChain p, ThreadPool piscina, String name) { AddRef (); DELETEME = false; occupato = false; //--> molto importante next = p; / / imposta il catenella - notare che questo è come una lista concatenata impl ThreadPool = piscina; / / imposta il pool di thread - Radice del ThreadPool ........ threadid = + + ThreadId; ...... / / avvia il filo thisThread = new Thread (this, nome + inttid.toString ()); thisThread.start (); }
Questa classe ha due metodi principali. Uno è booleano CanHandle () che è iniziato dal ThreadPool classe e poi procede in modo ricorsivo. Questo controlla se il thread corrente (attuale ThreadChain esempio) è libero di gestire il compito. Se è sta già gestendo un compito che chiama il successivo della catena.
pubblica booleana canHandle () {
if {/ / Se non occupato (occupato!)
System.out.println ("possibile gestire questo evento in id =" + threadid);
/ / todo segnalare un evento
try {
condLock.lock ();
condWait.signal () / / il segnale HandleRequest che è attesa per questo nel metodo run
.............................. ...........
return true;
}
................................... ......
/ / / Altrimenti se l'oggetto successivo della catena è libero
/ / / per gestire la richiesta
di ritorno next.canHandle ();Si noti che il HandleRequest è un metodo di ThreadChain che viene invocato da la corsa discussione () metodo e attende il segnale dal canHandle metodo. Si noti inoltre come l'attività viene gestita attraverso il pattern Command.
HandleRequest vuoto () {
.......
while (! CANCELLAMI) {
try {
............................
condLock. lock ();
condWait.await (); / / / Questo evento viene segnalato da
il MThd canHandle
condLock.unlock ();
....................... .....
starttime.set (System.currentTimeMillis ()); / / / Imposta
l'orario di inizio, AtomicLong - per verificare se il thread è appendere o meno
............... .............
/ / Ora set che questo oggetto come occupato, sta andando a gestire
una richiesta
occupato = true;
................ ............
Comando temp = threadpool.getRequest ();
m_timeout = temp.GetTimeOut () * 1000: / / / Il time out è espresso in secondi
temp.execute ();Ogni ThreadChain oggetto controlla anche se il thread ad esso associati è in uno stato appeso, in tal caso si imposterà ilDELETEME bandiera per scollegare dalla catena stessa.
HandleHungThreads void (ThreadChain prec) { ThreadChain next_p = GetNext (); boolean bIsHung = false; / / In pratica il ripristino dei collegamenti se uno della catena è appeso, simile a eliminare un nodo in una lista concatenata if (IsHung () | | ReleaseIdleThread ()) { bIsHung = true; se (questo == threadpool.root) / / / caso se root è appeso { threadpool.root = next_p; prev = next_p; } else { prev.next = GetNext () / / rimuove questa voce dal collegamento filo catena } } if (bIsHung) / / / se questo è un filo appeso che si sta svegliando { Release (); } ........... / / moltiplicazione la chiamata alla prossima nella catena if (bIsHung) { next.HandleHungThreads (prec); } else { next.HandleHungThreads (this); }
Si noti che aggiungendo o riducendo il thread è fatto in ThreadChain classe.
pubblica booleana canHandle () {
............................
if (occupato) / / / se il thread è già associato elaborazione di una richiesta
{
if (NULL == prossimo) {
/ / / significa che non c'è filo prossimo in modo da verificare se le discussioni dovrebbero essere aggiunto
/ / / se il numero di thread è già nulla da fare max
for (int i = (int) threadpool.m_threadCount ; i <= threadpool.maxthreadCount; + + i) {
next = nuovo ThreadChain (null, ThreadPool );/// creare un nuovo filo
break; / / / aggiungere un solo thread alla volta
.... ................................L'ultima classe qui è il ThreadPool classe. Questa agisce come una sorta di controllo anche se la maggior parte delle funzionalità di rilevare le discussioni appendere e aumentando o diminuendo il numero di thread in base al numero attuale di thread è fatto dal ThreadChain classe. Il ThreadPool detiene il contenitore con la coda compito e si avvia in fondo la catena per l'elaborazione.
pubblico ThreadPool (int _minthreadcount, int _maxthreadcount, maxtimeout lungo) {
.......
/ / Creazione del ThreadChain iniziale
/ / / C'è almeno un thread nel pool di thread
ThreadChain prev = new ThreadChain (null, questo " radice ");// / ultimo nodo
root = prev; / / / Inizialmente solo creare i fili minimo necessario for (int i = 0; i temp = ThreadChain nuovo ThreadChain (prec, questo); prev = temp; root = temp; }
I dettagli più fini riguardo a questa sono nel codice sorgente. Si noti che l'usanza ThreadPool non viene utilizzato in alcun codice di produzione e ho portato per divertimento. Quindi la sorgente fornito serve come un esempio solo
Strategie a confronto pool di thread nel contesto di attività Hanging
Questa sezione mette a confronto creazione dei thread e il comportamento rilasciando in base alla varie Java ThreadPoolstrategie discusse. Per testare ogni strategia Ho sparato un migliaio di attività, dopo che alcuni di loro in modo casuale impiccato. Dopo circa cinque secondi il numero di prestazioni eseguite è stato stampato.
Fisso ThreadPool strategia
ExecutorService piscina = Executors.newFixedThreadPool (100);- Totale attività avviate in Java ThreadPool = 482
- Totale attività completate in Java ThreadPool = 382
Con fili abbastanza questa strategia dovrebbe coprire il compito-hang problema, anche se il tasso di esecuzione dell'attività è piuttosto bassa. L'unico problema è che avete queste discussioni molti presenti indipendentemente dal numero di compiti, che è in testa.
Copia cache ThreadPool strategia
ExecutorService piscina = Executors.newCachedThreadPool ();- Totale attività avviate in Java ThreadPool = 1998
- Totale attività completate in Java ThreadPool = 1587
Si noti che alcuni compiti sono appesi che riflette il motivo per cui alcune discussioni sono ancora in corso. Qui il numero di thread pari al numero di attività che vengono licenziati. Se il tasso di attività è elevato, o in condizioni di scoppio, questo sarà immediatamente un problema: si può vedere il numero di thread di ripresa fino a migliaia di persone.
Altri ThreadPool Strategie
CallerRunsPolicy
ExecutorService piscina = ThreadPoolExecutor nuovi (0, 100, 250, TimeUnit.MILLISECONDS, nuove SynchronousQueue (), ThreadPoolExecutor.CallerRunsPolicy nuovo ());
Non utilizzare questo come il thread chiamante può bloccarsi.
BoundedQueue
ExecutorService piscina = ThreadPoolExecutor nuovo (10, 100, 250,
TimeUnit.MILLISECONDS, nuove ArrayBlockingQueue (10000)); - Totale attività avviate in Java ThreadPool = 38
- Totale attività completate in Java ThreadPool = 28
Una volta che tutti i fili appesi non più vengono creati.
Coda infinita
ExecutorService piscina = ThreadPoolExecutor nuovo (10, 1000, 250, TimeUnit.MILLISECONDS, nuove SynchronousQueue ()); - Totale attività avviate in Java ThreadPool = 1984
- Totale attività completate in Java ThreadPool = 1558
Assicurarsi di gestire l'eccezione compito respinta e anche mantenere il numero massimo di thread per un numero abbastanza grande per migliorare le prestazioni:
try { pool.execute (task4); } catch (RejectedExecutionException e) { .....
Personalizzati ThreadPool
ThreadPool tpool = new ThreadPool (2, 100, 1000);
- Totale attività iniziata nel CustomThreadPool = 1388
- Totale attività completata nel CustomThreadPool = 1109
Qui potete vedere il thread gestiti più attivamente e in maniera più intelligente e fili appesi sono individuati e rimossi.Speriamo che hai qualche idea da tutto questo!
Corso Java - Corsi Java - Corsi programmazione Java
Nessun commento:
Posta un commento
Nota. Solo i membri di questo blog possono postare un commento.