什么时候应该使用TaskCompletionSource <T>？

Question

AFAIK,它所知道的是,在某些时候,它SetResult或SetException方法被调用以Task<T>通过其Task属性完成暴露.

换句话说,它充当a Task<TResult>及其完成的生产者.

我在这里看到了这个例子:

如果我需要一种方法来异步执行Func并有一个Task来表示该操作.

public static Task<T> RunAsync<T>(Func<T> function) 
{ 
    if (function == null) throw new ArgumentNullException(“function”); 
    var tcs = new TaskCompletionSource<T>(); 
    ThreadPool.QueueUserWorkItem(_ => 
    { 
        try 
        {  
            T result = function(); 
            tcs.SetResult(result);  
        } 
        catch(Exception exc) { tcs.SetException(exc); } 
    }); 
    return tcs.Task; 
}

可以使用*如果我没有Task.Factory.StartNew- 但我确实有Task.Factory.StartNew.

题:

可有人请举例相关的情景解释直接到TaskCompletionSource 而不是一个假想中,我没有的情况 Task.Factory.StartNew？

Answer 1

当只有基于事件的api可用时(例如windows phone 8套接字),我主要使用它:

public Task<Args> SomeApiWrapper()
{
    TaskCompletionSource<Args> tcs = new TaskCompletionSource<Args>(); 

    var obj = new SomeApi();

    // will get raised, when the work is done
    obj.Done += (args) => 
    {
        // this will notify the caller 
        // of the SomeApiWrapper that 
        // the task just completed
        tcs.SetResult(args);
    }

    // start the work
    obj.Do();

    return tcs.Task;
}

因此,当与c#5 async关键字一起使用时,它特别有用.

Answer 2

根据我的经验,TaskCompletionSource将旧的异步模式包装到现代async/await模式中非常有用.

我能想到的最有益的例子就是工作时Socket.它具有旧的APM和EAP模式,但不具备和拥有的awaitable Task方法.TcpListenerTcpClient

我个人在NetworkStream课堂上有几个问题,而且更喜欢原始的Socket.因为我也喜欢这个async/await模式,我做了一个扩展类SocketExtender,它创建了几个扩展方法Socket.

所有这些方法都用于TaskCompletionSource<T>包装异步调用,如下所示:

    public static Task<Socket> AcceptAsync(this Socket socket)
    {
        if (socket == null)
            throw new ArgumentNullException("socket");

        var tcs = new TaskCompletionSource<Socket>();

        socket.BeginAccept(asyncResult =>
        {
            try
            {
                var s = asyncResult.AsyncState as Socket;
                var client = s.EndAccept(asyncResult);

                tcs.SetResult(client);
            }
            catch (Exception ex)
            {
                tcs.SetException(ex);
            }

        }, socket);

        return tcs.Task;
    }

我传递socket到BeginAccept方法,使我获得稍许的性能提升出来的没有扯起本地参数的编译器.

然后美丽的一切:

 var listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 listener.Bind(new IPEndPoint(IPAddress.Loopback, 2610));
 listener.Listen(10);

 var client = await listener.AcceptAsync();

Answer 3

对我来说,使用的经典场景TaskCompletionSource是,我的方法可能不一定需要耗费大量时间.它允许我们做的是选择我们想要使用新线程的特定情况.

一个很好的例子就是当你使用缓存时.您可以拥有一个GetResourceAsync方法,该方法在缓存中查找所请求的资源,并在找到资源时立即返回(不使用新线程,使用TaskCompletionSource).只有在找不到资源时,我们才会使用新线程并使用它来检索它Task.Run().

这里可以看到一个代码示例:如何使用任务有条不紊地运行代码

Answer 4

在这篇博客文章中,Levi Botelho描述了如何使用TaskCompletionSource为Process编写异步包装器,以便您可以启动它并等待其终止.

public static Task RunProcessAsync(string processPath)
{
    var tcs = new TaskCompletionSource<object>();
    var process = new Process
    {
        EnableRaisingEvents = true,
        StartInfo = new ProcessStartInfo(processPath)
        {
            RedirectStandardError = true,
            UseShellExecute = false
        }
    };
    process.Exited += (sender, args) =>
    {
        if (process.ExitCode != 0)
        {
            var errorMessage = process.StandardError.ReadToEnd();
            tcs.SetException(new InvalidOperationException("The process did not exit correctly. " +
                "The corresponding error message was: " + errorMessage));
        }
        else
        {
            tcs.SetResult(null);
        }
        process.Dispose();
    };
    process.Start();
    return tcs.Task;
}

及其用法

await RunProcessAsync("myexecutable.exe");

Answer 5

看起来没有人提到,但我猜单元测试也可以被认为是真实的生活.

我发现TaskCompletionSource在使用异步方法模拟依赖项时非常有用.

在实际测试中的程序:

public interface IEntityFacade
{
  Task<Entity> GetByIdAsync(string id);
}

在单元测试中:

// set up mock dependency (here with NSubstitute)

TaskCompletionSource<Entity> queryTaskDriver = new TaskCompletionSource<Entity>();

IEntityFacade entityFacade = Substitute.For<IEntityFacade>();

entityFacade.GetByIdAsync(Arg.Any<string>()).Returns(queryTaskDriver.Task);

// later on, in the "Act" phase

private void When_Task_Completes_Successfully()
{
  queryTaskDriver.SetResult(someExpectedEntity);
  // ...
}

private void When_Task_Gives_Error()
{
  queryTaskDriver.SetException(someExpectedException);
  // ...
}

毕竟,TaskCompletionSource的这种用法似乎是"一个不执行代码的Task对象"的另一种情况.

Answer 6

TaskCompletionSource用于创建不执行代码的Task对象.在真实世界场景中,TaskCompletionSource是I/O绑定操作的理想选择.这样,您可以获得任务的所有好处(例如,返回值,延续等),而不会在操作期间阻塞线程.如果您的"函数"是IO绑定操作,则不建议使用新任务阻止线程.而是使用TaskCompletionSource,您可以创建一个从属任务,以指示您的I/O绑定操作何时完成或出现故障.

Answer 7

这篇文章中有一个真实世界的例子,来自"Parallel Programming with .NET"博客.你真的应该阅读它,但无论如何这里是一个总结.

博客文章显示了两个实现:

"用于创建"延迟"任务的工厂方法,在发生某些用户提供的超时之前,实际上不会安排这些任务."

所示的第一个实现基于Task<>并且具有两个主要缺陷.第二个实施帖子继续通过使用缓解这些TaskCompletionSource<>.

这是第二个实现:

public static Task StartNewDelayed(int millisecondsDelay, Action action)
{
    // Validate arguments
    if (millisecondsDelay < 0)
        throw new ArgumentOutOfRangeException("millisecondsDelay");
    if (action == null) throw new ArgumentNullException("action");

    // Create a trigger used to start the task
    var tcs = new TaskCompletionSource<object>();

    // Start a timer that will trigger it
    var timer = new Timer(
        _ => tcs.SetResult(null), null, millisecondsDelay, Timeout.Infinite);

    // Create and return a task that will be scheduled when the trigger fires.
    return tcs.Task.ContinueWith(_ =>
    {
        timer.Dispose();
        action();
    });
}