多线程观察者模式

Question

我有一个问题，主题每次都会在不同的线程中更新。因此，每当主题更新时，它都会相应地使用新信息更新观察者。但是，如果观察者列表很长，则需要一些时间来更新所有观察者。想一想一个经常更新的主题。当主题正在更新观察者时，“主题”对象被锁定，因此不能由不同的线程更新。这将为主题创建信息流量或导致信息丢失。

您知道如何在多线程环境中处理这些问题吗？另外，有没有人能推荐一些关于C++并行编程的书？

Answer 1

考虑使用producer-consumer queues或message queues。在您的示例中，您可以通过两种方式使用队列：

1. 对该主题的更改将排队。当某些内容更新主题时，它会将新状态放入队列中并立即返回。这样，在通知观察者时，更新程序不会阻塞。您将需要一个不断将状态更改和对观察者的observers.
2. Notifications更新排入队列的线程。每个观察者都有一个发布主题状态更改通知的队列。

如果您使用的是Qt库，则可以将信号和插槽机制与Qt::QueuedConnection连接类型一起使用。该槽通过接收器的事件队列，并在接收器的线程中执行。这样，当接收器执行其各自的时隙时，发送器不会阻塞。

您的程序可能是Actor model (paradigm)的一个很好的候选者。以下是一些实现角色模型的C++库：

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

您的程序也可能是Dataflow范例的一个很好的候选者。查看建议的支持threading的Boost Dataflow库。

我没有一本书可以推荐，但可以看看Herb Sutter关于C++并发的series of Dr Dobbs articles。

Answer 2

我用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 <T extends AbstractListener> {

    interface Handler<T> {
        void handleEvent(T listener);
    }

    class BooleanHolder {
        boolean val;

        BooleanHolder(boolean v) {
            val = v;
        }

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

        boolean get() {
            return val;
        }
    }

    Map<AbstractListener, BooleanHolder> listeners = new HashMap<AbstractListener, BooleanHolder>();

    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<Entry<AbstractListener, BooleanHolder>> activeListeners = new ArrayList<Entry<AbstractListener, BooleanHolder>>();
        List<AbstractListener> inactiveListeners = new ArrayList<AbstractListener>(); 

        synchronized (listeners) {
            for (Entry<AbstractListener, BooleanHolder> entry : listeners.entrySet()) {
                if (entry.getValue().get()) {
                    activeListeners.add(entry);
                } else {
                    inactiveListeners.add(entry.getKey());
                }
            }
        }
         // call the method on active listener
        // 
        for (Entry<AbstractListener, BooleanHolder> 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);
            }
        }
    }
}