Silverlight 和 RX：为什么我必须调整浏览器的大小才能更新 UI？

Question

让我们从头开始：

我正在为 Silverlight 应用程序编写一个算法，它必须通过许多不同的高复杂性组合才能找到最佳值。为了让算法能够使用客户端上的所有给定资源，我决定提供一个并行版本。

首先，我编写了自己的面向异步事件的调度程序类，它带有一个等待句柄和一个阻塞对象，以限制并行线程的数量并在最后等待所有线程，直到我触发了最终的 CalculationCompletedEvent（顺便说一句：我正在使用 Backgroundworkers 来执行多线程）。但是有些东西不是线程安全的，结果列表中返回元素的数量不是恒定的。在同事向我指出 Reactive Extensions (rx) 之后，我考虑不要花更多时间来寻找泄漏。

为了了解如何使用它，我结合了consumer-producer example 和一些关于如何使用 rx 的建议（example1 和 example2）。

这很好用，但我不明白的是：为什么我必须调整浏览器的大小才能更新列表框并显示“_receivedStrings”的包含元素？又是一个小小的愚蠢疏忽？

顺便说一句：如果您不建议使用 rx，请试一试并告诉我为什么要使用其他方法。

XAML：

<UserControl x:Class="ReactiveTest.MainPage"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
    xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
    mc:Ignorable="d"
    d:DesignHeight="300" d:DesignWidth="400">

    <Grid x:Name="LayoutRoot" Background="White">
        <Grid.RowDefinitions>
            <RowDefinition/>
            <RowDefinition Height="Auto"/>
        </Grid.RowDefinitions>
        <ListBox HorizontalAlignment="Stretch" Name="listBox1" 
                 VerticalAlignment="Stretch" ItemsSource="{Binding}"/>
        <Button Grid.Row="1" Content="Klick me!" Width="Auto" Height="Auto" 
                HorizontalAlignment="Center" Click="Button_Click"/>
    </Grid>
</UserControl>

代码隐藏：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Animation;
using System.Windows.Shapes;
using System.IO;
using System.Reactive.Linq;

namespace ReactiveTest
{
    public partial class MainPage : UserControl
    {
        private int _parallelThreadsAmount;

        public IList<String> receivedStrings;

        public MainPage()
        {
            InitializeComponent();
            receivedStrings = 
                new List<String>();
            this._parallelThreadsAmount = 10;

            this.listBox1.DataContext = receivedStrings;
        }

        private void Button_Click(object sender, RoutedEventArgs e)
        {
            IList<IObservable<String>> obsCollection = 
                new List<IObservable<String>>();

            foreach (var item in forums)
            {
                obsCollection.Add(Calculate(item));
            }
            DateTime start = DateTime.Now;

            obsCollection.Merge(this._parallelThreadsAmount)
                .Subscribe(
                    y =>
                    {
                        receivedStrings.Add(
                            String.Format("{0} - Received: {1}", receivedStrings.Count, y));
                    },
                    () =>
                    {
                        DateTime end = DateTime.Now;
                        TimeSpan elapsed = end - start;
                        this.receivedStrings.Add(
                            String.Format(
                                "{0}/{1} done in {2} ms.", 
                                receivedStrings.Count, 
                                forums.Count(), 
                                elapsed.TotalSeconds)
                            );
                    }
                );
        }

        IObservable<String> Calculate(String source)
        {
            Random rand = new Random();
            return Observable.Defer(() => Observable.Start(() =>
            {
                // simulate some work, taking different time, 
                // to get the threads end in an other order than they've been started                
                System.Threading.Thread.Sleep(rand.Next(500, 2000));
                return source;
            }));
        }


        static readonly String[] forums = new string[]
        {
            "announce",
            "whatforum",
            "reportabug",
            "suggest",
            "Offtopic",
            "msdnsandbox",
            "netfxsetup",
            "netfxbcl",
            "wpf",
            "regexp",
            "msbuild",
            "netfxjscript",
            "clr",
            "netfxtoolsdev",
            "asmxandxml",
            "netfx64bit",
            "netfxremoting",
            "netfxnetcom",
            "MEFramework",
            "ncl",
            "wcf",
            "Geneva",
            "MSWinWebChart",
            "dublin",
            "oslo",
            // … some more elements
        };
    }
}

Answer 1

忽略缺乏 MVVM、IoC、可测试性等... 你没有实现 INotifyPropertyChanged，你没有使用 ObservableCollection(of T)。 1.将您的公共字段更改为公共只读属性 2. 使用 ObservableCollection 代替 IList

//public IList<String> receivedStrings; 
private readonly ObservableCollection<string> _receivedStrings = new ObservableCollection<string>();
public ObservableCollection<string> ReceivedStrings
{
    get { return _receivedStrings;}
}

您可能还必须使用 ObserveOnDispatcher() 来确保您在 Dispatcher 上被回调，因为您无法在不是 Dispatcher 线程的线程上更新 UI（即使通过 Binding）。

obsCollection.Merge(this._parallelThreadsAmount)                 
  .ObserveOn(Scheduler.Dispatcher)
  //-or-.ObserveOnDispatcher()
  //-or even better -.ObserveOn(_schedulerProvider.Dispatcher)
  .Subscribe(

Answer 2

不知何故，我想说你对 Rx 有点困惑，说实话这很正常 :)

正如 Lee Campbell 所写，对于初学者来说有几件事，在我看来，调整大小问题与 Rx 完全无关，而是 ObservableCollection 的东西。

除了我对你的代码进行了一些修改以处理我看到的一些事情之外，我不确定它是否正是你想要的，因为你有一个特定的任务来指定你想要的东西的并行度，但我会大胆地说这不是你关心的问题（希望我在这里不是一个聪明人）。

private void Button_Click(object sender, RoutedEventArgs e)
{
    //extension method in Rx for wrapping
    var obsCollection = forums.ToObservable();

    //Observing on the dispatcher to prevent x-thread exceptions
    obsCollection.Select( Calculate ).ObserverOnDispatcher().Subscribe(
        receivedString => { 
                receivedStrings.Add( String.Format("{0} - Received: {1}", receivedStrings.Count, y) );
            },
            ()=>{
                DateTime end = DateTime.Now;
                TimeSpan elapsed = end - start;
                this.receivedStrings.Add(
                    String.Format( "{0}/{1} done in {2} ms.", receivedStrings.Count, forums.Count(), elapsed.TotalSeconds)
                );
            }
    );
}

Random rand = new Random();
IObservable<string> Calculate(string inputString)
{
    //launching the "calculation" on the taskpool...can be changed to other schedulers
    return Observable.Start(
        ()=>{
            Thread.Sleep(rand.Next(150,250));

            return inputString;
            }, Scheduler.ThreadPool
        );
}