[JAVA] Observer-Observable (più Generics)

[tylerdurden83]

Ragazzi ho due domandine per voi :)

1) C'è un metodo della classe Method che si chiama invoke. Invoke ritorna un Object. Loro per farlo generico l'hanno fatto che ritorna Object, ma io chiamo uno specifico metodo da me scritto, che ritorna una List<String>. Poi passo questa List<String> come input ad un metodo scritto da me. Quindi:

List<String> lista = (List<String>)m.invoke(...);
miaClasse.doWork(lista);

se faccio così, ottendo un warning sulla prima delle due istruzioni. Se faccio:

List lista = (List)m.invoke(...);
miaClasse.doWork(lista);

non ottengo un warning sulla prima riga, ma lo prendo sulla seconda xke doWork ha nella signature una List<String> e non una List e basta. Naturalmente è solo un warning, però vorrei capire cosa mi sfugge, o se è effettivamente impossibile eliminarlo.

2) Domanda prettamente concettuale. Mettiamo che ho un file di testo, e un server che accetta richieste via socket. Quando un client richiede il file di stesto, il server apre un thread che apre uno stream sul file e manda via socket al client che l'ha richiesto. Siccome lo stream, prima di leggere la prima riga, lo posiziono alla fine del file di testo, alla fine tutti i client vedono arrivare le stesse info contemporaneamente (quindi non succede che un client sta alla 10 riga e uno alla 100sima, è un po come se piu terminali stessero in tail -f su un file di log movimentato da un'applicazione). Di conseguenza sarebbe buono se riuscissi a creare 1 solo thread e quindi 1 solo stream e a mandare il risultato della readline a tutti i client attualmente connessi. Qualora si connetterà un nuovo client, inizierò ad inviare anche a lui i risultati della readline, dal punto dove ero arrivato. Come posso implementarlo? Il pattern observer/observable mi può venire in aiuto?

Grazie 1000 come sempre,
Rob

[nuovoUtente86]

Se ho ben inteso vorresti avere un unico thread lettore e tanti thread che si occupano della comunicazione. In tal caso observer può venire sicuramente in aiuto, in modo da distribuire il multicast(dal lettore verso gli scrittori registrati) il testo letto.

[tylerdurden83]

Se ho ben inteso vorresti avere un unico thread lettore e tanti thread che si occupano della comunicazione. In tal caso observer può venire sicuramente in aiuto, in modo da distribuire il multicast(dal lettore verso gli scrittori registrati) il testo letto.

Prima di postare qualche spunto di implementazione vorrei levarmi un dubbio. Mi pare di aver capito che observable "passa" ai suoi observer il nuovo dato in serie, a seconda dell'ordine di registrazione, e finchè un observer non conferma la ricezione del dato, non lo passa agli altri observer, corretto? Vediamo se ritrovo il link da cui ho preso tale info...(trovato, eccolo (http://forums.sun.com/thread.jspa?threadID=5248890))

PS: se quello che ho scritto sopra è corretto, c'è il pericolo che, ad esempio, ho 2 thread ascoltatori, leggo un nuovo dato, lo comunico ad entrambi i thread, il primo lo prende ma non lo manda subito, il secondo ancora non lo prende, nel frattempo l'observable sovrascrive il dato letto prima con uno nuovo, il primo thread observer finisce di elaborare il primo dato, al secondo observer arrivano tutti e due i dati letti, oppure solo il secondo?

[nuovoUtente86]

no il dato è inviato,concettualmente, a tutti i client contemporaneamente e non prevede riscontro, ma c'è l' overhead della chiamata al metodo update. Eventualmente puoi farne una tua versione customizzata.

[tylerdurden83]

no il dato è inviato,concettualmente, a tutti i client contemporaneamente e non prevede riscontro, ma c'è l' overhead della chiamata al metodo update. Eventualmente puoi farne una tua versione customizzata.

Dal link che ho postato mi sembrava di capire che lui registra due observer, uno Slow con una sleep di 10 secondi dentro (registrato per primo), e uno Normal, senza sleep, e che Normal consumava il dato solo dopo 10 secondi (cioè quando il primo terminava) e non immediatamente come avrebbe potuto fare. Sbaglio?

[nuovoUtente86]

come ti ho scritto su concettualmente puoi considerare la notifica immediata, ma nelle pratica devi considerare il tempo consumato dal metodo update.
Devi essere tu a decidere, e quindi ottimizzare tale metodo. Alla fin fine a te serve ricevere il dato, ci sarà poi il thread che gestisce il socket di riferimento a consegnarlo sulla rete.

[banryu79]

Dal link che ho postato mi sembrava di capire che lui registra due observer, uno Slow con una sleep di 10 secondi dentro (registrato per primo), e uno Normal, senza sleep, e che Normal consumava il dato solo dopo 10 secondi (cioè quando il primo terminava) e non immediatamente come avrebbe potuto fare. Sbaglio?
No, hai capito bene.

Il fatto è che il metodo update() di un Observer viene invocato dal metodo notifyObservers() di Observable; se dentro quest'ultimo metodo i vari Observer.update() vengono chiamati dallo stesso thread, è chiaro che si verifica, tra la chiamata dell' update() di un Observer e l'altro, un lag temporale pari al tempo neccessario che il thread deve impiegare nell'eseguire il codice della specifica implementazione di update().

Chiaro che quindi non è una "furbata" chiamare una Thread.sleep() dentro l'update() di un Observer :D

Al limite, sapendo di avere questa neccessità, uno (ad esempio) quando estende Observable fa l'ovverride del metodo notifyObservers() in modo da usare un thread per ogni chiamata ad Observer.update(), cioè un thread per ogni Observer da notificare.

Per curiosità posto lo spezzone di codice che illustra il metodo Observable.notifyObservers() nel JDK:

...
public void notifyObservers(Object arg) {
/*
* a temporary array buffer, used as a snapshot of the state of
* current Observers.
*/
Object[] arrLocal;

synchronized (this) {
/* We don't want the Observer doing callbacks into
* arbitrary code while holding its own Monitor.
* The code where we extract each Observable from
* the Vector and store the state of the Observer
* needs synchronization, but notifying observers
* does not (should not). The worst result of any
* potential race-condition here is that:
* 1) a newly-added Observer will miss a
* notification in progress
* 2) a recently unregistered Observer will be
* wrongly notified when it doesn't care
*/
if (!changed)
return;
arrLocal = obs.toArray();
clearChanged();
}

for (int i = arrLocal.length-1; i>=0; i--)
((Observer)arrLocal[i]).update(this, arg);
}
...

(Nota: 'obs' è di tipo Vector e contiene tutti gli Observer registrati)

@EDIT:
(Nota 2: in particolare, pare che basarsi sull'odine di inserimento degli Observer, come suggeriva qualcuno nel thread che hai linkato, sembrerebbe una pessima idea: la documentazione di Observable dice chiaramente che l'ordine con cui vengono invocati gli Observers è "unspecified".
E infatti questi vengono conservati in un Vector, dove vengono inseriti con addElement(), il che significa che ogni nuovo Observer inserito viene inserito in coda al Vector, ma poi, come puoi vedere nel codice qui sopra, il vector stesso viene, molto evangelicamente, ciclato al contrario, della serie: gli ultimi saranno i primi).

[nuovoUtente86]

Poichè l' override di notifyObservers comporta la definizione di una nuova struttura di storage per gli observer (obs è private), si può pensare (modificando di conseguenza gli add e delete) anche a strutture eventualmente più performanti e/o a ordinare in base a delle discriminanti in fase si inserimento.
Ma di norma tutto può essere risolto lato observer, senza modifiche al subject.

[tylerdurden83]

Ottimo grazie a tutti e due, mi metto a studiare un po', poi magari posto qualcosa se ancora non mi sarà tutto chiaro.

Per curiosità, che mi dite della domanda sui generics? So che è solo un warning, ma sono curioso di sapere se non sono riuscito io a toglierlo, oppure è effettivamente irrisolvibile...

[nuovoUtente86]

Ottimo grazie a tutti e due, mi metto a studiare un po', poi magari posto qualcosa se ancora non mi sarà tutto chiaro.

Per curiosità, che mi dite della domanda sui generics? So che è solo un warning, ma sono curioso di sapere se non sono riuscito io a toglierlo, oppure è effettivamente irrisolvibile...

http://java.sun.com/docs/books/tutorial/java/generics/erasure.html

[tylerdurden83]

Ho fatto una prova al volo:


public class Main {
public static void main(String[] args) {
ReaderThread readerThread = new ReaderThread();
ConsumerThread output = new ConsumerThread();
ConsumerThread output2 = new ConsumerThread();
ConsumerThread output3 = new ConsumerThread();
readerThread.addObserver(output);
readerThread.addObserver(output2);
readerThread.addObserver(output3);
new Thread(readerThread).start();
}
}

public class ReaderThread extends Observable implements Runnable{
@Override public void run(){
try {
FileInputStream fis = new FileInputStream("C:/Users/Roberto/Documents/NetBeansProjects/Streamer/log.txt");
InputStreamReader isr = new InputStreamReader(fis);
BufferedReader br = new BufferedReader(isr);
String str = "";
while(true){
if((str = br.readLine()) != null ){
System.out.println("ReaderThread: "+str);
setChanged();
notifyObservers(str);
}
Thread.sleep(1000L);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}

public class ConsumerThread implements Observer {
public void update(Observable ob, Object o) {
if (o instanceof String) {
final String text = (String) o;
new Thread(new Runnable(){
public void run() {
try {
if(Thread.currentThread().getName().equalsIgnoreCase("Thread-1")){
Thread.sleep(4000L);
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
System.out.println("ConsumerThread received "+text);
}
}).start();
}
}
}


L'output è stato:

ReaderThread: a
ConsumerThread received a
ConsumerThread received a
ReaderThread: b
ConsumerThread received b
ConsumerThread received b
ConsumerThread received b
ReaderThread: c
ConsumerThread received c
ConsumerThread received c
ConsumerThread received c
ReaderThread: d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received a
ReaderThread: e
ConsumerThread received e
ConsumerThread received e
ConsumerThread received e
ReaderThread: f
ConsumerThread received f
ConsumerThread received f
ConsumerThread received f
ReaderThread: g
ConsumerThread received g
ConsumerThread received g
ConsumerThread received g

La sequenzialità non è garantita, ad esempio il thread-1 ha stampato (nel caso delle socket, inviato) prima b che a, ma mi va bene anche così. Grazie a tutti! (ps la cosa dei generics non l'ho capita benissimo, li parla di runtime, io ho un warning in compilazione che non capisco se impossibile da levare oppure no...)

[banryu79]

Ho fatto una prova al volo:
[...snip...]
L'output è stato:

ReaderThread: a
ConsumerThread received a
ConsumerThread received a
ReaderThread: b
ConsumerThread received b
ConsumerThread received b
ConsumerThread received b
ReaderThread: c
ConsumerThread received c
ConsumerThread received c
ConsumerThread received c
ReaderThread: d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received a
ReaderThread: e
ConsumerThread received e
ConsumerThread received e
ConsumerThread received e
ReaderThread: f
ConsumerThread received f
ConsumerThread received f
ConsumerThread received f
ReaderThread: g
ConsumerThread received g
ConsumerThread received g
ConsumerThread received g

La sequenzialità non è garantita, ad esempio il thread-1 ha stampato (nel caso delle socket, inviato) prima b che a, ma mi va bene anche così.

Perdona, ma non capisco cosa hai provato: nel senso che tu ne codice fai eseguire al Thread-1, e solo a lui, una sleep di 4 sec... Chiaro che l'ordine in cui ti stampa l'output è quello, nel codice che hai postato i tre thread dei tre diversi update non sono sincronizzati tra loro...

Se togli quel controllo sul Thread-1 per la sleep, l'output verrà stampato nell'ordine che ti aspetti!

Ho paura di essermi perso qualcosa, vero?

@EDIT: per capirci meglio:
se tu modifichi CosumerThread in modo da fargli comunque stampare un messaggio, anche nel caso dia il Thread-1, prima di metterlo in sleep, vedrai che la "sequenzialità", è rispettata e garantita:
Modifica:

class ConsumerThread implements Observer
{
@Override public void update(Observable ob, Object o) {
if (o instanceof String) {
System.out.println(Thread.currentThread().getName());
final String text = (String) o;
final Thread updateExecutor = new Thread(new Runnable() {
@Override public void run() {
try {
if (Thread.currentThread().getName().equalsIgnoreCase("Thread-1")) {
System.out.println("sleep required on Consumer Thread received "+text);
Thread.sleep(4000L);
}
} catch (InterruptedException ex) { ex.printStackTrace(); }
System.out.println("ConsumerThread received "+text);
}
});
updateExecutor.start();
}
}
}


Output:

run:
ReaderThread: a
sleep required on Consumer Thread received a *e va in sleep*
ConsumerThread received a
ConsumerThread received a
ReaderThread: b
ConsumerThread received b
ConsumerThread received b
ConsumerThread received b
ReaderThread: c
ConsumerThread received c
ConsumerThread received c
ConsumerThread received c
ReaderThread: d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received d
ConsumerThread received a

[nuovoUtente86]

vedrai che la "sequenzialità", è rispettata e garantita
quando parliamo di thread dati in pasto allo scheduler, senza alcuno strumento di sincronizzazione non si può fare alcuna assunzione su cosa e in che ordine avvenga. Che poi nel caso in esame funzioni, per via della sleep sul producer, è un altro discorso ma non ci si può affidare al caso o a una sleep. Poi è anche vero che dal dominio del problema mi sembra di capire che l' unica cosa da garentire sia l' ordine in lettura e non c'è nulla da sincronizzare tra i vari thread che operano separatamente sui singoli socket.

[tylerdurden83]

La sleep era puramente a scopo di test, non ci sarà nessuna cosa del genere nella versione finale. Quello che mi interessava controllare con quel semplice esempio era quanto segue, partendo dal fatto che l'invio di a da sulla socket (in questo caso una print) da parte di uno dei tre observer ha avuto un delay:

ReaderThread: a
Thread-2 ConsumerThread received a
Thread-3 ConsumerThread received a
ReaderThread: b
Thread-4 ConsumerThread received b
Thread-5 ConsumerThread received b
Thread-6 ConsumerThread received b
ReaderThread: c
Thread-7 ConsumerThread received c
Thread-8 ConsumerThread received c
Thread-9 ConsumerThread received c
ReaderThread: d
Thread-10 ConsumerThread received d
Thread-11 ConsumerThread received d
Thread-12 ConsumerThread received d
ReaderThread: e
Thread-1 ConsumerThread received a
Thread-14 ConsumerThread received e
Thread-15 ConsumerThread received e
Thread-13 ConsumerThread received e

1- L'eventuale delay sull'invio di a da parte di uno degli observer non comporta rallentamenti in lettura/invio di b etc a nessun observer(tantomeno l'observer stesso che ha avuto problemi ad inviare a)
2- L'observer che deve ancora inviare a non si perde nessun update prima dell'invio di a

Nuovo esempio(sempre puramente di test):


public class ConsumerThread implements Observer {
private int id;
ConsumerThread(int id){
this.id=id;
}
public void update(Observable ob, Object o) {
if (o instanceof String) {
final String text = (String) o;
//new Thread(new Runnable(){
// public void run() {
try {
int sonno=0;
while(sonno<5){
if(this.id==1){
Thread.sleep(1000L);
System.out.println(id+" ha dormito 1 sec...");
sonno++;
} else {
break;
}
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
System.out.println(this.id+" ConsumerThread received "+text);
//}
//}).start();
}
}
}


Output:

ReaderThread: a
3 ConsumerThread received a
2 ConsumerThread received a
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ConsumerThread received a
ReaderThread: b
3 ConsumerThread received b
2 ConsumerThread received b
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ha dormito 1 sec...
1 ConsumerThread received b
ReaderThread: c
3 ConsumerThread received c
2 ConsumerThread received c
1 ha dormito 1 sec...

In questo caso, immaginiamo che l'invio sulla socket non è fatto su un thread separato ma direttamente dentro l'update dell'observer. La sequenzialità è garantita, ma se uno degli observer ci mette un pochino ad inviare il dato(e potrebbe benissimo essere così nel mio caso), si ferma tutto, il thread Reader non va avanti e tantomeno gli altri observer).

Quindi mi si sono delineati due scenari:
Caso 1 -> invio di ogni variabile letta dentro un nuovo thread
+ Se un dato ci mette tempo ad essere inviato il reader va avanti e tutti gli observer continuano a ricevere i nuovi dati
- I dati non arrivano necessariamente in ordine sequenziale

Caso 2 -> invio direttamente da update senza aprire un nuovo thread per farlo
+ I dati arrivano in ordine
- Se update di un observer si inchioda per vari secondi, si ferma tutto

Forse c'è qualcosa che mi avete detto e m'è scappato, spero di no però...

[banryu79]

Quindi mi si sono delineati due scenari:
Caso 1 -> invio di ogni variabile letta dentro un nuovo thread
+ Se un dato ci mette tempo ad essere inviato il reader va avanti e tutti gli observer continuano a ricevere i nuovi dati
- I dati non arrivano necessariamente in ordine sequenziale

Ok,forse potresti dotare ogni client-consumer registrato di
- una struttura concorrente opportuna (vedi package java.util.concurrent) dove accumulare ogni singola lettura del thread-lettore (penso a una coda di priorità)
- un thread che la svuota, sparando le print sulla socket

Vado a naso però, non ho alcuna esperienza in materia.
Penso dipenda anche da quello che stai cercando di implementare (stavo immaginando un server che produce continuamente dei dati di log, da una parte, e dall'altra dei client che si connettono e da quel momento ricevono le letture del server in modo sequenziale, finchè non si disconnettono. In tale caso ci sono altri dettagli a cui pensare: ad esempio l'uso di una thread pool per i client [a meno che non hai la garanzia che siano sempre quattro gatti] e le sue politiche [java.util.concurrent provvede anche a questo]).

[tylerdurden83]

Abuserò un pochino della tua conoscenza...

Ho usato

static ExecutorService service = Executors.newCachedThreadPool();

condiviso tra tutti i thread consumatori. Quindi invece di


new Thread(new Runnable(){
public void run() {
try {
if(Thread.currentThread().getName().equalsIgnoreCase("Thread-1")){
Thread.sleep(4000L);
}
} catch (InterruptedException ex) {
Logger.getLogger(ConsumerThread.class.getName()).log(Level.SEVERE, null, ex);
}
System.out.println("ConsumerThread received "+text);
}
}).start();


ora uso:


SenderThread senderThread = new SenderThread(text);
service.execute(senderThread);


Dove SenderThread spara sulla socket. Una curiosità però, essendo ExecutorService static, vari metodi chiameranno service.execute(Runnable) contemporaneamente sullo stesso oggetto. Immagino che service.execute(Runnable) sia thread safe, tuttavia vedendo il codice di ThreadExecutorPool che lo implementa:


public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}


non noto nulla che mi lasci pensare che sia thread safe... cosa mi sono perso?

[banryu79]

Abuserò un pochino della tua conoscenza...

Ahahah... allora caschi male :asd:
E pensare che ti avevo anche avvertito :Prrr:


non noto nulla che mi lasci pensare che sia thread safe... cosa mi sono perso?
Beh, 'workQueue' è una java.util.concurrent.BlockingQueue...
Poi fa uso di varibili volatile, ove oppourtono, e dei meccanismi di locking implementati a livello di API nel package stesso (il corpo del metodo 'ensureQueuedTaskHandled' è controllato da un java.util.concurrent.ReentrantLock).

Comunque qua chi ti può dare consigli pratici con cognizione di causa, secondo me, è PGI. Io ancora non ho intrapreso uno studio serio e approfondito di tutti i meccanismi di sincronizzazione della tecnologia Java (primitive del linguaggio + API per la concorrenza), per non parlare dello studio completo del Java Memory Model.

Se vuoi farti un'idea di massima, tanto per gradire, leggi questo tutorial (http://java.sun.com/docs/books/tutorial/essential/concurrency/index.html).

[nuovoUtente86]

ExecutorService è tread safe perchè fa uso di BlockingQueue.
Quanto al tuo progetto la situazione è abbastanza lineare.
- Un thread lettoro Observable
- N (quanti sono dinamicamente i client connessi) thread, observer, che wrappano i socket e si occupano della comunicazione.
- Cosa devono avere di peculiari tali Thread: si devono sincronizzare, con il lettore, su una condizione (ovvero la disponibilità di un nuovo dato letto).
Questo lo puoi fare sfruttando le condition messe a disposizione dal framework oppure te la implementi da te.
- Il metodo Update, infine, cosa deve fare? Deve semplicemente ricevere il dato e metterlo a disposizione del thread scrittore: operazione, di fatto, a costo costante.

[tylerdurden83]

E' vero che la collezione BlockingQueue è thread safe, tuttavia se un metodo fa delle operazioni tipo:

if (poolSize >= corePoolSize ....

e queste non sono sincronizzate, avrei detto che non è thread safe. Ad es se un metodo fa:

if i = 0 then print "OK"

potrei avere la print di OK anche se nel frattempo i non vale più 0, perchè le istruzioni non sono nè atomiche nè sincronizzate. Probabilmente però è come dite voi, e basta che tutte le variabili presenti nelle istruzioni condizionali siano volatile (unito ad oggetti controllati da lock) per renderlo thread safe, anche se avrei detto che nel caso di if (a==0 || b==0), anche se a e b sono volatile, quando passo a==0 e vado a valutare b (che valuto correttamente perchè è volatile) a potrebbe essere cambiata...

[nuovoUtente86]

Se vai a spulciare i sorgenti dell' intera classe, ti farai una panoramica più ampia sui meccanismi di sincronizzazione utilizzati.

[tylerdurden83]

Se vai a spulciare i sorgenti dell' intera classe, ti farai una panoramica più ampia sui meccanismi di sincronizzazione utilizzati.

Me la sto spulciando, tuttavia già semplicemente:


public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {


da quello che leggo qui (http://www.ibm.com/developerworks/java/library/j-jtp06197.html):


...
Locks offer two primary features: mutual exclusion and visibility.
...
Volatile variables
Volatile variables share the visibility features of synchronized, but none of the atomicity features. This means that threads will automatically see the most up-to-date value for volatile variables. They can be used to provide thread safety, but only in a very restricted set of cases: those that do not impose constraints between multiple variables or between a variable's current value and its future values. So volatile alone is not strong enough to implement a counter, a mutex, or any class that has invariants that relate multiple variables (such as "start <=end").


ne deduco che sebbene volatile garantisca l'accesso al valore più recente, è irrilevante quando si tratta di fare i++ oppure if(i==0 && j==0), volatili o meno non fa differenza in quel caso. Insomma, i++ o if(i==0 && ...) se non stanno in un metodo sincronizzato o in un blocco synchronized, non vedo come possano essere thread safe, indipendentemente dal resto della classe... :muro:

[banryu79]

E' vero che la collezione BlockingQueue è thread safe, tuttavia se un metodo fa delle operazioni tipo:

if (poolSize >= corePoolSize ....

e queste non sono sincronizzate, avrei detto che non è thread safe.

Nota che in 'ensureQueuedTaskHandled' poolSize è volatile e dentro il metodo è solo letta, non modificata.
(Inoltre, se arriva a chiamare questo metodo vuol dire che o poolSize è >= di coorPoolsize oppure il motodo 'addIfUnderCorePoolSize' a restituito false).

Se vai a spulciarti la cosa noterai che l'unico punto in cui poolSize viene modificata è in un metodo privato della classe, e questo metodo privato viene chiamato internamente assicurando la thread-safety in scrittura perchè prima viene acquisito il mainLock (un ReentrantLock usato internamente dalla classe).

Purtroppo non riesco a entrare più nel dettaglio e nel merito, perchè, come ti dicevo prima, mi manca sia l'esperienza in materia che un solido studio della teoria (sia di quella generale che di quella specifica per Java).

Comunque, se è per rassicurarti, ti è sufficiente leggere la documentazione della classe, che ti illustra le garanzie offerte.
Per informarsi bene purtroppo non ci sono (che io sappia) dei tutorial specifici sul sito della Sun, l'unica è comperare un buon libro sull'argomento e studirsi le specifiche del memory model.

Di veloce & facile consultazione trovi degli articoli della IBM: http://www.ibm.com/developerworks/views/global/libraryview.jsp?site_id=1&contentarea_by=Java%20technology&sort_by=Type&sort_order=2&start=1&end=31&topic_by=Multithreading&product_by=-1&type_by=All%20Types&show_abstract=true&search_by=

Alcuni però sono molto :old:

@EDIT:
Rettifico e sottolineo: la conoscenza delle specifiche del Memory Model è un requisito imprescindibile per comprendere davvero fino in fondo il multithreading in Java.

[tylerdurden83]

Vado di implementazione concreta va, vediamo se funzionerà ehehe

Grazie mille del preziosissimo aiuto!!

[asso7]

Ciao!
Mi scuso se mi intrometto nella discussione.... ma in pratica sto implementando una cosa molto simile a questa di cui discutete...
però non capisco come posso gestire i socket e l'observer/observable CONTEMPORANEAMENTE!
immaginate di avere tanti client observer e 1 server observable.... come dovrei gestire il tutto?

[tylerdurden83]

Ciao!
Mi scuso se mi intrometto nella discussione.... ma in pratica sto implementando una cosa molto simile a questa di cui discutete...
però non capisco come posso gestire i socket e l'observer/observable CONTEMPORANEAMENTE!
immaginate di avere tanti client observer e 1 server observable.... come dovrei gestire il tutto?

Ti dico cosa faccio io nel mio caso, spero ti possa essere utile.
1) quando starto il server questo prende da un file di configurazione una lista di log di interesse, e su ognuno di quelli crea un

class LogReader extends Observable implements Runnable

che sta in tail -f diciamo, e ogni volta che legge un'entry chiama setChanged(); e notifyObservers(lineRead);

Fin qui nulla di speciale penso.
2) Quei runnable sono startati e gestiti da un LogReadersManager. Questo ha una mappa interna di associazioni <String,LogReader>. Anche qui nulla d speciale.
3) Il server riceve una request, e crea un nuovo thread per gestirla.

public class RequestHandler implements Observer, Runnable {

RequestHandler apre i vari stream e riceve dal client una request per un certo log. Se la richiesta ha una corrispondenza, chiama LogReadersManager.registerObserver(this, flow);

public static void registerObserver(Observer o, String flow){
this.logMap.get(flow).addObserver(o);
}

quindi si registra a ricevere gli update del log richiesto dall'utente.

[asso7]

umh... ma se io avessi un vettore che deve essere usato in modalità produttore/consumatore (cioè il client inserisce gli elementi e un manager da la possibilità ad un server di estrarre tali elementi)... ha senso usare un meccanismo di observer per notificare coda piena/coda vuota? (so che potrei usare semplicemente gli stream... ma vorrei una implementazione migliore.... )