简单的 Rx 示例要么阻塞要么有竞争条件

Question

观看后：Intro to Reactive Programming我正在尝试使用 C# 和reactiveX 复制示例。总之，它是一个货币转换器。用户指定他们想要的货币数量和类型。当前汇率是从服务器检索的，返回所需的时间不定。由于这个可变的服务器时间，旧的货币请求可能会在最近的请求之后进入。这个警告被用作使事情具有反应性以避免竞争条件错误的动机。我可以通过在代表货币汇率的任务上使用 Wait() 来避免我的反应代码中的竞争条件，但这会阻止 GUI。我应该如何接近这个？也许我应该取消其他任务？下面是我的代码，它不会阻塞但存在竞争条件问题。public CurrencyWindow(){    InitializeComponent();    var amount = Observable.FromEventPattern<TextChangedEventArgs>(        txtAmount, "TextChanged").Select(t=>int.Parse(txtAmount.Text));    var yen = Observable.FromEventPattern<RoutedEventArgs>(        rdoYen, "Checked").Select(t => "Yen");    var dollar = Observable.FromEventPattern<RoutedEventArgs>(        rdoDollar, "Checked").Select(t => "Dollar");    var pound = Observable.FromEventPattern<RoutedEventArgs>(        rdoPound, "Checked").Select(t => "Pound");    // merge the currency selection into one stream.    var currencyType = yen.Merge(dollar).Merge(pound);    // Lets now get a stream of currency rates.    var exchangeRate = currencyType.Select(s => GetExchangeRate(s));    amount.CombineLatest(exchangeRate, async (i, d) => i * await d).Subscribe(async v =>    {        var val = await v;        txtCost.Text = val.ToString();    });}private Task<double> GetExchangeRate(string currency){    double exchange = 0;    switch(currency)    {        case "Yen":            exchange = 1;            break;        case "Pound":            exchange = 2;            break;        case "Dollar":            exchange = 3;            break;    }    return Task.Factory.StartNew(() =>     {        System.Threading.Thread.Sleep((int)exchange * 3000);        return exchange;    });}

Answer 1

试试这个代码，看看它是否有效：

var exchangeRate = currencyType

    .Select(s => Observable.FromAsync(() => GetExchangeRate(s)))

    .Switch();

amount

    .CombineLatest(exchangeRate, (i, d) => i * d)

    .Subscribe(v =>

    {

        var val = v;

        txtCost.Text = val.ToString();

    });

此代码将确保只有最新currencyType生成的才会返回汇率。如果出现了一种新货币，那么所有先前的飞行调用都GetExchangeRate将被忽略（即使它们在当前调用之后出现）。