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

Question

让我们从头开始：

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

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

为了了解如何使用它，我结合了 消费者-生产者示例有关如何使用 rx 的一些建议（example1 和<一href="http://social.msdn.microsoft.com/Forums/en-US/rx/thread/655b80f6-f487-42a0-a5c8-18222ed95f54" rel="nofollow">示例2）。

这很好用，但我不明白的是：为什么我必须调整浏览器大小才能更新列表框并显示“_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
        );
}