是否应该在单独的线程中通知观察者？答案

【问题标题】：Should Observers be notified in separate threads each one?是否应该在单独的线程中通知观察者？
【发布时间】：2012-11-24 13:33:30
【问题描述】：

我知道这听起来很重，但我正在尝试解决一个假设的情况。想象一下，您有 N 个对象的观察者。每个人都对对象状态感兴趣。当应用Observer Pattern 时，observable 对象倾向于遍历其观察者列表，调用观察者notify()|update() 方法。

现在想象一个特定的观察者对 observable 对象的状态有很多工作要做。例如，这会减慢最后一次通知的速度。

因此，为了避免减慢向所有观察者的通知速度，我们可以做的一件事是在单独的线程中通知观察者。为了让它工作，我想每个观察者都需要一个线程。为了避免因繁重的工作而导致通知速度变慢，这是我们正在承受的痛苦开销。如果使用线程方法，比放慢速度更糟糕的是无限循环导致的死线程。阅读有经验的程序员对于这个来说会很棒。

在设计问题上有多年经验的人是怎么想的？
这是一个没有实质性解决方案的问题吗？
真的是个坏主意吗？为什么？

示例

这是一个模糊的例子，目的是为了演示并希望澄清我什至没有测试过的基本想法：

class Observable(object):
    def __init__(self):
        self.queues = {}

    def addObserver(self, observer):
        if not observer in self.queues:
            self.queues[observer] = Queue()
            ot = ObserverThread(observer, self.queues[observer])
            ot.start()

    def removeObserver(self, observer):
        if observer in self.queues:
            self.queues[observer].put('die')
            del self.queues[observer]

    def notifyObservers(self, state):
        for queue in self.queues.values():
            queue.put(state)

class ObserverThread(Thread):
    def __init__(self, observer, queue):
        self.observer = observer
        self.queue = queue

    def run(self):
        running = True
        while running:
            state = self.queue.get()
            if state == 'die':
                running = False
            else:
                self.observer.stateChanged(state)

【问题讨论】：

标签： multithreading observer-pattern

【解决方案1】：

你在正确的轨道上。

每个观察者拥有自己的输入队列和自己的消息处理线程是很常见的（或者更好的是：队列拥有线程，观察者拥有队列）。见Active object pattern。

但是有一些陷阱：

如果您有 100 或 1000 名观察员，您可能需要使用 thread pool pattern
请注意，您将失去对事件处理顺序的控制（哪个观察者首先处理事件）。这可能不是问题，或者可能会打开一个包含非常难以检测的错误的潘多拉盒子。这取决于您的具体应用。
您可能必须处理观察者在通知者之前被删除的情况。这可能有点难以正确处理。
您需要实现消息而不是调用函数。消息生成可能需要更多资源，因为您可能需要分配内存、复制对象等。您甚至可能希望通过为常见消息类型实现消息池来进行优化（您也可以选择实现包装此类池的消息工厂)。
为了进一步优化，您可能希望生成一条消息并将其发送给所有观察者（而不是生成同一消息的多个副本）。您可能需要为您的消息使用一些reference counting 机制。

【讨论】：

Observable 只是通知观察者并让他们在他们认为合适的时候采取行动不是更好吗？
@NarendraPathai：这导致了一种更复杂的模式：每个观察者都必须实现一种方法来获知事件，并实现另一种方法来处理事件（在其自己的内部线程中）。此外，即使在事件中做出重要决定，在某些情况下也可能需要大量资源。
@LiorKogan 在我的例子中，观察者首先处理通知的顺序并不重要。但是，必须保证观察者接收到的变化与 observable 及时变化的顺序相同，并且他们不能错过任何变化。但是有一个问题：如果每个观察者都有自己的线程来顺序处理通知，为什么要避免调用观察者方法而使用消息呢？
@Sebastian：当然你必须调用观察者方法，但是你需要为这些方法传递参数。消息正是可以排队的一组参数。消息类型的数量等于您必须调用的方法的数量。如果只有一种方法会更简单，就像您的情况一样：状态更改方法，但请考虑有多种消息类型/方法的情况，例如：EventAHappened、EventBHappened 等 - 每个事件都有自己的参数集。
@LiorKogan 我认为更好和可扩展的方法是避免为不同的消息类型创建不同的方法。应该广播消息，每个观察者都应该知道如何处理它。如果您以为自定义消息编写自定义方法而告终，那么您将以需要为新类型的消息更改所需的复杂维护代码而告终。只有一种通知方法，您不需要知道观察者将如何处理消息，也不需要添加方法调用。

【解决方案2】：

让每个观察者自己决定它的反应是否是重量级的，如果是，则启动一个线程，或者向线程池提交一个任务。在单独的线程中发出通知并不是一个好的解决方案：在释放可观察对象的同时，它限制了单线程通知的处理器能力。如果您不信任观察者，则创建一个线程池，并为每个通知创建一个任务并将其提交到池中。

【讨论】：

我明白了，但是，如果观察者不知道处理更新需要多长时间怎么办？

【解决方案3】：

在我看来，当您为 Observable 拥有大量的 Observers 时，它们会进行繁重的处理，那么最好的办法是在 Observer 中使用 notify() 方法。

notify() 的使用：只需将Observer 中的dirty 标志设置为true。因此，只要 Observer 线程发现它合适，它就会查询 Observable 以获取所需的更新。

这不需要在Observable 端进行繁重的处理并将负载转移到Observer 端。

现在取决于Observers 何时必须观察。

【讨论】：

@Narenda 我认为这种方法的主要问题是观察者可能会丢失可观察对象中的一些更改，因为观察者没有收到更改的“快照”，而是被告知更改已被做了，最终会查询到对象的状态。因此，当观察者进行查询时，可观察对象可能会再次发生变化。结果是我们丢失的对象发生了变化。
@Sebastian 在你的论点之后我觉得这取决于应用程序的类型（每个快照是否关键）。如果是 RSS 阅读器，我认为我的解决方案应该更好！

【解决方案4】：

@Pathai 的答案在很多情况下都是有效的。

一个是您正在观察数据库中的变化。在许多方面，您无法仅从快照中重建最终状态，尤其是当您的状态作为复杂查询从数据库中获取时，并且快照是对数据库的更新。

要实现它，我建议使用 Event 对象：

class Observer:
    def __init__(self):
        self.event = threading.Event()

# in observer:
while self.event.wait():
    # do something
    self.event.clear()

# in observable:
observer.event.set()

【讨论】：