订阅可观察对象（主题）的响应式扩展

Question

我只是第一次在 winforms 应用程序中使用 Reactive Extensions。请注意，过去 4 年我一直在做 Web 开发，并且我非常熟悉淘汰赛中的可观察对象和可观察模式，我猜这会导致我在这里感到困惑。

无论如何，对于问题和代码。我正在构建一个简单的 winforms 实验（见下文）来说明我的问题。直到 start new 中的线程完成后，下面的订阅才会运行。我可以跟踪对 OnNext 的调用，但订阅直到有时 20-30 秒后才会触发。有人可以向我解释这种行为吗？

public partial class Form1 : Form
{

    private Subject<int> progress;
    private CancellationToken cancellationToken;
    private IScheduler _scheduler;

    public Form1()
    {
        InitializeComponent();

        CancellationTokenSource source = new CancellationTokenSource();
        cancellationToken = source.Token;

        _scheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);


    }

    private void Start_Click(object sender, EventArgs e)
    {
        progress
            .ObserveOn(_scheduler)
            //.Throttle(TimeSpan.FromSeconds(5))
            .Subscribe(
                (i) => {
                    progressBar1.Do<ProgressBar>(ctl =>
                    {
                        ctl.Value = i;
                    });
                },
                (ex) => { },
            cancellationToken
        );

        Task counterTask = Task.Factory.StartNew(() =>
        {
            for (var i = 1; i < 101; i++)
            {
                Thread.Sleep(500);
                progress.OnNext(i);
            }
        }, cancellationToken,
        TaskCreationOptions.LongRunning,
        TaskScheduler.FromCurrentSynchronizationContext()
      );
    }

    private void Form1_Load(object sender, EventArgs e)
    {
       progress = new Subject<int>();
    }


}



public static class ControlExtensions
{
    public static void Do<TControl>(this TControl control, Action<TControl> action)
      where TControl : Control
    {
        if (control.InvokeRequired)
            control.Invoke(action, control);
        else
            action(control);
    }
}

Answer 1

您的问题来自您的任务在 UI 线程上运行的事实，因为您使用的是TaskScheduler.FromCurrentSynchronizationContext()。

因此，您的各种 Sleep 调用阻塞了 UI 线程，冻结了 UI（例如，无法拖动窗口）并阻止您的可观察订阅执行（因为 ObserveOn，它应该在 UI 线程上执行调度程序）。

将TaskScheduler.FromCurrentSynchronizationContext() 替换为TaskScheduler.Default（后台TaskPool 线程），一切都会如您所愿。

请注意，您无需调用Do/Invoke，因为您已经在您提供的调度程序的 UI 线程上。