单个生产者多个消费者的架构[关闭]答案

【问题标题】：architecture for single producer multiple consumers [closed]单个生产者多个消费者的架构[关闭]
【发布时间】：2013-05-23 05:15:03
【问题描述】：

我正在设计一个 GUI 来显示市场数据。目前，我有多个使用相同数据但显示不同方面的窗口。我正在尝试找出构建此系统的最佳方法。

我有一个市场数据生产者和多个向生产者注册回调的消费者。当生产者准备好数据时，它会遍历消费者列表并通过回调将数据分发给每个消费者。我不确定这是否是分发数据的最佳方法。每个消费者必须等待前一个消费者完成回调处理才能获取其数据。

有没有一种方法可以让所有消费者同时或以最小的延迟获取数据？我正在使用 C# 4.0，想知道是否有任何语言功能可以实现这一点。

【问题讨论】：

对于单一答案来说太宽泛了。投票结束。
尽量减少问题并用一些代码说明。
你想错了。消费者必须相互等待的要求解决了这个问题。您确实没有有 N 个消费者，每个消费者执行 1 次操作，您有 1 个消费者执行 N 次操作。这是微不足道的。

标签： c# c#-4.0 architecture

【解决方案1】：

为什么不使用Reactive Extensions Framework？它应该是你所描述的想法，它提供了很大的灵活性。这里是如何管理multiple subscriptions

【讨论】：

如果应用程序正在运行所有 InProcess RX 确实是一个不错的选择
你应该在你的答案中包含一些代码 - 现在它几乎只是链接。

【解决方案2】：

您可以通过为每个消费者使用单独的BlockingCollection 来寻求一个相当简单的解决方案。

生产者线程只需要它自己管理的BlockingCollection 队列中的List<>。当生产者线程产生一个项目时，它只需使用BlockingCollection.Add() 将其添加到列表中的每个队列中。注意：这是相同的项目被多次排队 - 每个消费者队列一次。

项目类型应该是不可变的引用类型 - 然后在消费者之间共享是安全的，并且开销只是每个消费者队列中的引用。

这不是最有效的路线，因为它需要每个消费者排队，但我不认为这太糟糕，除非排队很大。不过，它很容易理解和实施。

BlockingCollection 可以很容易地告诉消费者线程何时没有更多数据，以便他们可以干净地退出。

生产者线程调用BlockingCollection.CompleteAdding() 向消费线程发出信号，告知他们应该在没有更多数据时退出。

同时，所有消费线程需要做的就是这个（假设queue是它们的BlockingCollection队列）：

foreach (var item in queue.GetConsumingEnumerable())
{
    ... process item
}

就是这样。当没有更多数据时，消费者将自动阻塞在foreach 循环中，当没有更多数据时，他们将自动退出foreach并且生产者调用了CompleteAdding()。

【讨论】：

【解决方案3】：

对于这种情况，您应该使用 RabbitMQ 或 MSMQ 等队列。它们是此类发布者/订阅者操作的最佳选择。

如果这不是一个选项，您可以使用线程触发回调或将它们设置为 WCF 触发并忘记调用 (OneWay)。

【讨论】：

【解决方案4】：

如果从它们自己的线程（很可能是线程池线程）调用每个注册的回调很重要，那么这样做很容易：

public class Foo
{
    private HashSet<Action> callbacks = new HashSet<Action>();
    public event Action MyEvent
    {
        add
        {
            callbacks.Add(value);
        }
        remove
        {
            callbacks.Remove(value);
        }
    }

    private void FireMyEvent()
    {
        foreach (var action in callbacks)
        {
            ThreadPool.QueueUserWorkItem(o => { action(); });
        }
    }
}

【讨论】：