multithread Observer modello

Question

Ho una domanda in cui un soggetto viene aggiornato in un thread diverso ogni volta. Quindi, ogni volta che il soggetto viene aggiornato, aggiorna corrispondentemente l'osservatore con le nuove informazioni. Tuttavia, se l'elenco degli osservatori è lungo, sarà necessario del tempo per aggiornare tutti gli osservatori. Pensa a un argomento che viene aggiornato molto frequentemente. Mentre il soggetto sta aggiornando gli osservatori, l'oggetto "oggetto" è bloccato e quindi non può essere aggiornato da un thread diverso. Ciò creerà traffico di informazioni per oggetto o causerà la perdita di informazioni.

Avete qualche idea di come questi problemi vengono gestiti in un ambiente multi-threaded? Inoltre, qualcuno può consigliare alcuni libri sulla programmazione parallela con C++?

Answer 1

considerare l'uso di producer-consumer queues o message queues. Per il tuo esempio, puoi utilizzare una coda in due modi:

1. Le modifiche all'oggetto sono in coda. Quando qualcosa aggiorna l'oggetto, inserisce il nuovo stato nella coda e restituisce immediatamente. In questo modo, l'updater non si blocca mentre gli osservatori vengono avvisati. Avrai bisogno di un thread che annulli continuamente le modifiche di stato e aggiorni gli osservatori.

2. notifiche agli osservatori sono in coda. Ogni osservatore ha una coda in cui vengono inviate notifiche di cambio stato dell'oggetto.

Se si utilizza la libreria Qt, è possibile utilizzare il segnale & slot meccanismo con il tipo di connessione Qt::QueuedConnection. Lo slot passa attraverso la coda degli eventi del ricevitore ed è eseguito nel thread del destinatario. In questo modo, il mittente non blocca mentre i ricevitori eseguono i loro rispettivi slot.

tuo programma potrebbe essere un buon candidato per il Actor model (paradigm). Qui ci sono alcune librerie C++ che implementano il modello attore:

• Theron
• libcppa (C++ 11 based)
• Asynchronous Agents Library (Microsoft)

Il vostro programma potrebbe anche essere un buon candidato per il paradigma Dataflow. Controlla la libreria Boost Dataflow proposta, che supporta threading.

---

Non ho un libro di raccomandare, ma guarda che la concorrenza di series of Dr Dobbs articles su C++ Herb Sutter.

Answer 2

ho scritto un pattern Observer multithreading in Java

 

    import java.lang.reflect.Method; 
    import java.util.List; 
    import java.util.Map; 
    import java.util.Map.Entry; 

    /** 
    * An observer pattern that allows listeners to register(), unregister() in 
    * multiple threads and also notify listeners in another thread. 
    * 
    * A HashMap keeps track of the listeners and their status (active, obsolete). 
    * When a listener unregister, its entry is marked as obsolete in this map. 
    * 
    * During firing of an event, the observer notifies all the listeners that are 
    * active, the active status will be stored in a Boolean that's synchronized so 
    * rare race conditions like calling notify on an active listener that has just 
    * turned obsolete will not happen. 
    * 
    * 
    */ 
    public class MultithreadedObserverPattern { 

     interface Handler { 
      void handleEvent(T listener); 
     } 

     class BooleanHolder { 
      boolean val; 

      BooleanHolder(boolean v) { 
       val = v; 
      } 

      void set(boolean v) { 
       val = v; 
      } 

      boolean get() { 
       return val; 
      } 
     } 

     Map listeners = new HashMap(); 

     public void register(AbstractListener l) { 
      synchronized (listeners) { 
       listeners.put(l, new BooleanHolder(true)); 
      } 
     } 

     public void unregister(AbstractListener l) { 
      synchronized (listeners) { 
       BooleanHolder status = listeners.get(l); 
       if (status != null) { 
        // notify call also syncing on status 
        synchronized (status) { 
         status.set(false); 
        } 
       } 
       // set to false 
      } 
     } 

     public void notifyAll(Handler handler) { 
      // here we do not synchroznie on listeners to avoid tricky lock situations 
      // make a copy of the map 
      List> activeListeners = new ArrayList>(); 
      List inactiveListeners = new ArrayList(); 

      synchronized (listeners) { 
       for (Entry entry : listeners.entrySet()) { 
        if (entry.getValue().get()) { 
         activeListeners.add(entry); 
        } else { 
         inactiveListeners.add(entry.getKey()); 
        } 
       } 
      } 
      // call the method on active listener 
      // 
      for (Entry e : activeListeners) { 
       BooleanHolder status = e.getValue(); 
       // remove those listeners that are no longer active 
       synchronized (status) { 
        if (status.get()) { 
          handler.handleEvent(e.getKey()); 
        } 
       } 
      } 

      synchronized (listeners) { 
       // remove inactive listeners 
       for (AbstractListener l : inactiveListeners) { 
        listeners.remove(l); 
       } 
      } 
     } 
    }