标签: system.reactive

微软在.NET Framework 4中引入了BCL IObservable<T>接口,我想,"太好了,最后,我必须使用它!" 所以我深入挖掘并阅读帖子和文档,甚至实现了模式.

在这样做之后,我意识到基本实现实际上将所有T事件发送给所有订阅者,而不对其进行任何过滤; 即普通广播.我在某处读到了Observable模式用于普通广播.我觉得这不是真的,我错过了一些东西.

我的问题:

1. 如果我添加一个过滤机制,使用Observable模式和只使用普通CLR事件有什么区别？

2. 应该何时使用此模式,何时应该选择使用普通CLR事件？

3. Observable模式的主要优点是什么？

SignalR和Reactive Extensions一样吗？你能解释为什么或为什么不解释？

什么时候会选择使用Rx而不是TPL,或者2个框架是否正交？

根据我的理解,Rx主要用于提供事件的抽象并允许组合,但它也允许提供异步操作的抽象.使用Createxx重载和Fromxxx重载并通过处理返回的IDisposable取消.

TPL还通过任务和取消功能提供操作抽象.

我的困境是什么时候使用哪种情况？

我希望使用c#在Windows窗体应用程序中按分钟(按时钟)每分钟做一些事情.我只是想知道最好的方法是什么？

我可以使用一个计时器并将其间隔设置为60000,但为了让它在一分钟内运行,我必须准确地启用它,而不是真的可行.

我可以使用一个计时器并将其间隔设置为1000.然后在其tick事件中,我可以根据我设置的变量检查当前时钟,如果分钟已经改变,则运行我的代码.这让我很担心,因为我让我的电脑每1秒做一次检查,以便每1分钟进行一次工作.当然这很难看？

我正在使用Windows窗体和.Net 2.0,因此不想使用.Net 3.5附带的DispatchTimer

这一定是一个相当普遍的问题.你有没有更好的方法来做到这一点？

我想知道新的C#5.0异步功能对于Rx - Reactive Extensions意味着什么？它似乎不是替代品,但它们似乎重叠 - Task而且IObservable.

MSDN上的这个页面说明了这一点

如果不使用以调度程序作为参数的重载,Rx将使用最小并发原则选择默认调度程序.这意味着选择引入满足运算符需求的最少并发性的调度程序.例如,对于返回具有有限和少量消息的observable的运算符,Rx调用Immediate.对于返回潜在大量或无限数量消息的运算符,将调用CurrentThread.对于使用计时器的操作员,使用ThreadPool.

我想实际上有一个参考表,可观察的操作员使用哪个默认的Scheduler,但我找不到任何地方.每个可观察运营商的默认调度程序是什么？

Reactive Extensions带有许多辅助方法,用于将现有事件和异步操作转换为可观察对象,但是如何从头开始实现IObservable <T>？

IEnumerable有一个可爱的yield关键字,使其实现起来非常简单.

实现IObservable <T>的正确方法是什么？

我需要担心线程安全吗？

我知道有人支持在特定的同步上下文中回调,但这是否是我作为IObservable <T>作者需要担心或者以某种方式内置的内容？

更新:

这是Brian的F#解决方案的C#版本

using System;
using System.Linq;
using Microsoft.FSharp.Collections;

namespace Jesperll
{
    class Observable<T> : IObservable<T>, IDisposable where T : EventArgs
    {
        private FSharpMap<int, IObserver<T>> subscribers = 
                 FSharpMap<int, IObserver<T>>.Empty;
        private readonly object thisLock = new object();
        private int key;
        private bool isDisposed;

        public void Dispose()
        {
            Dispose(true);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing && !isDisposed)
            {
                OnCompleted();
                isDisposed = true;
            }
        }

        protected void OnNext(T value)
        {
            if (isDisposed) …

编辑:感谢wiki的链接,我认为自从它已经开始,它更容易去那里检查.不过这里的问题也很好,所以不在msdn论坛周围的人会了解wiki及其位置.

简短问题:

你有一个Rx代码示例,可以帮助人们更好地理解它吗？

漫长的漫无边际的问题:

现在已经发布了Rx框架,我认为我们中的许多人都有兴趣获取这些内容并尝试它们.可悲的是,实际上并没有很多例子(经过详尽的搜索后,我几乎确信Rx只是为了在wpf应用上轻松实现).

我不记得我读到或听过的确切位置(我一直在看很多博客和视频),Rx团队似乎对101系列感兴趣...当他们有足够的时间去做...对于那些想要了解它并且现在玩它的人来说非常糟糕(我的意思是,当一个像这样的新技术出现时,自我尊重的开发人员不会觉得自己是一个带有新玩具的孩子) .

我个人现在一直在尝试,但哇有一些疯狂的概念......只是让像Materialise和Zip这样的方法让我想起了Teleporters和Back to the Future的东西.

所以,我认为这将是很好,如果有更多的了解的,帮助建立的范例集,阿拉101个LINQ的例子,从基本的使用变为更复杂的东西,几乎涵盖了所有的方法和他们的用途,一种实用的方法(也许还有一点理论,特别是因为这些概念可能需要它)

我认为MS开发人员花时间给我们这样的材料很棒,但我也认为这个社区足以开始构建我们自己的材料,不是吗？

有人可以解释Observable和ConnectableObservable之间的区别吗？Rx Extensions文档非常稀疏,我不明白ConnectableObservable在什么情况下是有用的.

此类用于Replay/Prune方法.

鉴于Reactive Extensions(Rx)框架提供的可组合事件的好处,我想知道我的类是否应该停止推送.NET事件,而是暴露Rx可观察量.

例如,使用标准.NET事件获取以下类:

public class Foo
{
   private int progress;
   public event EventHandler ProgressChanged;

   public int Progress
   {
      get { return this.progress; }
      set
      {
         if (this.progress != value)
         {
            this.progress = value;

            // Raise the event while checking for no subscribers and preventing unsubscription race condition.
            var progressChanged = this.ProgressChanged;
            if (progressChanged != null)
            {
                progressChanged(this, new EventArgs());
            }
         }
      }
   }
}

很多单调的管道.

这个类可以使用某种observable来替换这个功能:

public class Foo
{
   public Foo()
   {
       this.Progress = some new observable; …