将ConcurrentQueue<T>作为IObservable<T>公开？

Question

我想知道是否可以将队列（特别是ConcurrentQueue）用作IObservable的源？就像这样：

  Queue = new ConcurrentQueue<IMessage>();
  var xs = Queue.AsEnumerable().ToObservable();

  xs.Subscribe((IMessage msg) =>
     {
        Console.WriteLine("Msg :" + msg.subject);
     });

我猜这并没有太多意义，因为没有任何东西被出队。我正在尝试实现一个非阻塞的进程，可以订阅推送到观察者的“消息”，因此使用了队列。我相信我应该能够用RX做到这一点，但似乎无法理解！

如果您有任何关于如何实现这一点的建议，我会感兴趣。谢谢！

- Jason Hyland

Answer 1

你是正确的，将一个队列（并发或简单，没有关系）转换只会枚举它，但不会出队。“真正”的实现是可能的，但更复杂-请参见我在RX论坛上提出的类似问题的链接（与StackOverflow相比，这仍然是更好的RX信息来源）： 如何使用RX实现单个工作程序消费者生产者队列？

- Sergey Aldoukhov

---

谢谢Sergey，确认这不是我所希望的那么轻松 :) 我在RX论坛上找到了你最初的问题，OneByOneTask<T>演示了'Observable.GenerateInSequence'语法。看起来很不错！ - Jason Hyland