Patrón de observador multiproceso

Question

Tengo una pregunta donde un tema se actualiza en un hilo diferente cada vez. Por lo tanto, cada vez que se actualiza el tema, actualiza al observador con la nueva información. Sin embargo, si la lista de observadores es larga, requerirá algún tiempo para actualizar a todos los observadores. Piense en un tema que se actualiza con mucha frecuencia. Mientras el sujeto está actualizando los observadores, el objeto "sujeto" está bloqueado y, por lo tanto, no se puede actualizar con un hilo diferente. Esto creará tráfico de información para el sujeto o causará la pérdida de información.

¿Tiene alguna idea de cómo se manejan estos problemas en un entorno de subprocesos múltiples? Además, ¿alguien puede recomendar algunos libros sobre programación paralela con C++?

Answer 1

Considere el uso de producer-consumer queues o message queues. Para su ejemplo, puede usar una cola de dos formas:

1. Los cambios en el Asunto están en cola. Cuando algo actualiza el tema, pone el nuevo estado en la cola y lo devuelve inmediatamente. De esta forma, el actualizador no se bloquea mientras se notifica a los observadores. Necesitará un hilo que deque continuamente los cambios de estado y actualice los observadores.

2. Las notificaciones a los observadores están en cola. Cada observador tiene una cola donde se publican las notificaciones de cambio de estado del sujeto.

Si está utilizando la biblioteca Qt, se puede utilizar el mecanismo de señales & ranuras con el tipo Qt::QueuedConnection conexión. La ranura pasa por la cola de eventos del receptor y se ejecuta en el hilo del receptor. De esta forma, el emisor no bloquea mientras los receptores ejecutan sus respectivas ranuras.

Su programa puede ser un buen candidato para el Actor model (paradigm). Aquí hay algunas bibliotecas de C++ que implementan el modelo de actor:

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

Su programa podría también ser un buen candidato para el paradigma Dataflow. Consulte la biblioteca Boost Dataflow propuesta, que admite threading.

---

que no tienen un libro para recomendar, pero echa un vistazo a la concurrencia de series of Dr Dobbs articles en C++ Herb Sutter.

Answer 2

me escribió un patrón de observador en Java multiproceso

 

    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); 
       } 
      } 
     } 
    }