如何获取调度程序的 TaskScheduler？

Question

我有一个具有多个 Dispatcher（又名 GUI 线程、又名消息泵）的应用程序，以确保 GUI 的缓慢且无响应的部分运行，而不会严重影响应用程序的其余部分。我也经常使用 Task。

目前，我的代码可以在 TaskScheduler 或 Dispatcher 上有条件地运行 Action，然后返回 Task直接或使用 TaskCompletionSource 手动创建一个。然而，这种人格分裂的设计使得处理取消、例外等事情比我想象的要复杂得多。我想在任何地方使用 Task 而在任何地方使用 DispatcherOperation。为此，我需要在调度程序上安排任务 - 但如何做呢？

如何获取任何给定 Dispatcher 的 TaskScheduler？

编辑：经过以下讨论，我确定了以下实现：

public static Task<TaskScheduler> GetScheduler(Dispatcher d) {
    var schedulerResult = new TaskCompletionSource<TaskScheduler>();
    d.BeginInvoke(() => 
        schedulerResult.SetResult(
            TaskScheduler.FromCurrentSynchronizationContext()));
    return schedulerResult.Task;
}

Answer 1

步骤 1：创建扩展方法：

public static Task<TaskScheduler> ToTaskSchedulerAsync (
    this Dispatcher dispatcher,
    DispatcherPriority priority = DispatcherPriority.Normal) {

    var taskCompletionSource = new TaskCompletionSource<TaskScheduler> ();
    var invocation = dispatcher.BeginInvoke (new Action (() =>
        taskCompletionSource.SetResult (
            TaskScheduler.FromCurrentSynchronizationContext ())), priority);

    invocation.Aborted += (s, e) =>
        taskCompletionSource.SetCanceled ();

    return taskCompletionSource.Task;
}

步骤 2：使用扩展方法：

旧语法：

var taskSchedulerAsync = Dispatcher.CurrentDispatcher.ToTaskSchedulerAsync ();
var taskFactoryAsync = taskSchedulerAsync.ContinueWith<TaskFactory> (_ =>
    new TaskFactory (taskSchedulerAsync.Result), TaskContinuationOptions.OnlyOnRanToCompletion);
// this is the only blocking statement, not needed once we have await
var taskFactory = taskFactoryAsync.Result;
var task = taskFactory.StartNew (() => { ... });

新语法：

var taskScheduler = await Dispatcher.CurrentDispatcher.ToTaskSchedulerAsync ();
var taskFactory = new TaskFactory (taskScheduler);
var task = taskFactory.StartNew (() => { ... });

Answer 2

不幸的是，没有内置的方法可以做到这一点。没有专门用于在 TaskScheduler 中包装 Dispatcher 的内置类 - 我们拥有的最接近的类是包装 SynchronizationContext 的类。从 SynchronizationContext 构建 TaskScheduler 的唯一公共 API 是 Paul Michalik 提到的：TaskScheduler.FromCurrentSynchronizationContext - 正如您所观察到的，仅当您已经在相关同步上下文中（即在相关调度程序的线程上）时才有效。

因此，您有三个选择：

1. 安排您的代码，以便需要相关调度程序的调度程序的类在某个时刻有机会在这些调度程序的线程上运行，以便您可以使用 TaskScheduler.FromCurrentSynchronizationContext正如预期的那样。
2. 使用 Dispatcher.BeginInvoke 在调度程序线程上运行一些代码，并在该代码中调用 TaskScheduler.FromCurrentSynchronizationContext。 （换句话说，如果你不能安排 1. 自然发生，那就强迫它发生。）
3. 编写你自己的任务调度程序。

Answer 3

查看TaskScheduler.FromCurrentSynchronizationContext。即使应用程序强加了特定的线程模型，任务框架也提供了一种非常灵活的方法来配置计算绑定操作的执行。

编辑：

嗯，很难从您发布的内容中得到更明确的信息。我知道您正在运行某种多视图应用程序，每个视图都有单独的调度程序，对吧？由于所有调度都归结为获取 SynchronizationContext 和 Post - 您可以获取正确的 TaskScheduler （具有正确 TaskScheduler 的任务） code>SynchronizationContext) 在您的视图获得同步上下文的某个时刻。一个简单的方法是在配置任务期间获取 TaskScheduler：

 // somewhere on GUI thread you wish to invoke
 // a long running operation which returns an Int32 and posts
 // its result in a control accessible via this.Text
 (new Task<Int32>(DoSomeAsyncOperationReturningInt32)
      .ContinueWith(tTask => this.Text = tTask.Result.ToString(),
                    TaskScheduler.FromCurrentSynchronizationContext)).Start();

不确定这是否有帮助，如果您广泛使用任务，您可能已经知道......

Answer 4

您可以将整个函数写在一行中：

public static Task<TaskScheduler> ToTaskSchedulerAsync(this Dispatcher dispatcher,
                           DispatcherPriority priority = DispatcherPriority.Normal)
{
    return dispatcher.InvokeAsync<TaskScheduler>(() =>
         TaskScheduler.FromCurrentSynchronizationContext(), priority).Task;
}

那些对默认 UI 线程感到满意的人可能会发现以下内容足以满足需要：

var ts = Application.Current.Dispatcher.Invoke<TaskScheduler>(() => TaskScheduler.FromCurrentSynchronizationContext());

Answer 5

尽管没有专门用于将 Dispatcher 包装在 TaskScheduler 中的内置类，但您可以自己编写一个类。完成该工作的示例实现如下所示：

internal class DispatcherTaskScheduler : TaskScheduler
{
    private readonly Dispatcher _dispatcher;
    private readonly SendOrPostCallback _dispatcherCallback;

    public DispatcherTaskScheduler(Dispatcher dispatcher)
    {
        _dispatcher = dispatcher;
        // Callback for dispatcher.
        _dispatcherCallback = (state) =>
        {
            _ = TryExecuteTask((Task)state!);
        };
    }

    public override int MaximumConcurrencyLevel
    {
        // Dispatcher operates on one thread only.
        get => 1;
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        // Whole method is only for debugging purposes. No need to implement it.
        return Array.Empty<Task>();
    }

    protected override void QueueTask(Task task)
    {
        // Schedule the task for execution.
        _ = _dispatcher.BeginInvoke(DispatcherPriority.Normal, _dispatcherCallback, task);
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        // If we are in the right thread then execute the task.
        if (_dispatcher.CheckAccess())
            return TryExecuteTask(task);
        return false;
    }
}

Answer 6

这就是我总是将异步函数调用转换为同步函数调用的方式（向互联网上的某人致敬）：

    public static class ThreadingUtils 
    {
         public static TaskScheduler GetScheduler(Dispatcher dispatcher)
         {
             using (var waiter = new ManualResetEvent(false))
             {
                 TaskScheduler scheduler = null;
                 dispatcher.BeginInvoke(new Action(() =>
                 {
                     scheduler = 
                         TaskScheduler.FromCurrentSynchronizationContext();
                     waiter.Set();
                 }));
                 waiter.WaitOne();
                 return scheduler;
             }
         }
    }

变化：

    if (!waiter.WaitOne(2000))
    { 
        //Timeout connecting to server, log and exit
    }