c# - IScheduler.Schedule 与 IScheduler.ScheduleAsync？

Question

IScheduler接口(interface)提供

public static IDisposable Schedule(this IScheduler scheduler, Action action)

和

public static IDisposable ScheduleAsync(this IScheduler scheduler, Func<IScheduler, CancellationToken, System.Threading.Tasks.Task<IDisposable>> action)

ScheduleAsync 的方法说明:

    // Summary:
    //     Schedules work using an asynchronous method, allowing for cooperative scheduling
    //     in an imperative coding style.
    //
    // Parameters:
    //   scheduler:
    //     Scheduler to schedule work on.
    //
    //   action:
    //     Asynchronous method to run the work, using Yield and Sleep operations for
    //     cooperative scheduling and injection of cancellation points.
    //
    // Returns:
    //     Disposable object that allows to cancel outstanding work on cooperative cancellation
    //     points or through the cancellation token passed to the asynchronous method.
    //
    // Exceptions:
    //   System.ArgumentNullException:
    //     scheduler or action is null.

有人可以解释这两种方法之间的区别吗？

我什么时候应该使用 ScheduleAsync？

我什么时候应该使用时间表？

允许以命令式编码风格进行协作调度意味着什么？

谢谢。

Answer 1

转发

此答案基于直接来自 Rx 团队的明确解释 this帖子 - 注意它很长并且涵盖的不仅仅是这一点。转到标题为在 Rx 查询运算符中利用“异步” 的部分，所有内容都得到了解释，包括标题为使调度程序更易于执行的部分中关于 ScheduleAsyc 的特定示例与“等待”一起使用

这是我试图解释的:

总结

ScheduleAsync 的主要动机是采用 C# 5 的异步/等待功能来简化编写代码，以对许多事件执行“公平”调度，否则可能会导致其他操作的调度程序饥饿。这就是“合作调度”的意思——与共享调度程序的其他代码一起玩得很好。为此，您可以安排下一个事件，然后交出控制权直到该事件触发并挂接到该事件以安排您的下一个事件，依此类推。

在 Rx 2.0 之前，这是通过递归调度实现的。

简单的例子

这是链接文章中的示例，它提供了范围运算符的实现。这个实现很差，因为它不让出控制权而使调度程序挨饿:

static IObservable<int> Range(int start, int count, IScheduler scheduler)
{
    return Observable.Create<int>(observer =>
    {
        return scheduler.Schedule(() =>
        {
            for (int i = 0; i < count; i++)
            {
                Console.WriteLine("Iteration {0}", i);
                observer.OnNext(start + i);
            }
            observer.OnCompleted();
        });
    });
}

请注意 OnNext 如何处于一个循环中，在不放弃控制的情况下敲打调度程序(如果调度程序是单线程的，则尤其糟糕)。它使其他操作没有机会安排它们的操作，并且不允许在取消时中止。我们如何解决这个问题？

递归调度 - Pre-Rx 2.0 解决方案

这是用递归调度解决这个问题的老方法——很难看出发生了什么。这不是“命令式编码风格”。递归调用 self() 在您第一次看到它时非常不透明 - 在我的例子中是第十次，尽管我最终得到了它。 This classic post by the legendary Bart de Smet will tell you more than you'll ever need to know about this technique .无论如何，这是递归风格:

static IObservable<int> Range(int start, int count, IScheduler scheduler)
{
    return Observable.Create<int>(observer =>
    {
        return scheduler.Schedule(0, (i, self) =>
        {
            if (i < count)
            {
                Console.WriteLine("Iteration {0}", i);
                observer.OnNext(start + i);
                self(i + 1); /* Here is the recursive call */
            }
            else
            {
                observer.OnCompleted();
            }
        });
    });
}

为了更公平，如果订阅被处置，下一个待定的预定操作将被取消。

新的 Async/await 样式

这是通过 async/await 的编译器转换实现延续的新方法，它允许“命令式编码风格”。请注意，与递归样式相比，其动机更易读 - async/await 脱颖而出，通常以 .NET 惯用的方式显示正在发生的事情:

static IObservable<int> Range(int start, int count, IScheduler scheduler)
{
    return Observable.Create<int>(observer =>
    {
        return scheduler.ScheduleAsync(async (ctrl, ct) =>
        {
            for (int i = 0; i < count; i++)
            {
                Console.WriteLine("Iteration {0}", i);
                observer.OnNext(i);
                await ctrl.Yield(); /* Use a task continuation to schedule next event */
            }
            observer.OnCompleted();

            return Disposable.Empty;
        });
    });
}

它看起来就像一个 for 循环，但实际上 await ctrl.Yield() 将让出控制权以允许其他代码进入调度程序。它使用 Task continuations 一次只安排一个事件——也就是说，每次迭代仅在前一个迭代完成后才发布到调度程序，从而避免直接在调度程序上排长队。取消也有效，这次 Rx 框架将订阅的处置转换为通过 ct 传入的取消 token 。

我推荐阅读 original post如果链接仍然有效，我就把它拿走了!