.net core Task的学习理解(下)

• .net core Task的学习理解(下)
  • Task.GetResult与死锁
  • SynchronizationContext与Task回调处理
    • 自定义SynchronizationContext
  • ValueTask

Task.GetResult与死锁

通过下面的代码，可能会造成死锁的情况。

static async Task Main(string[] args)
{
    Console.WriteLine("Test:Start");
    DoSthAsync().GetAwaiter().GetResult();
    Console.WriteLine("Test:END");
    Console.ReadLine();
}
static async Task DoSthAsync()
{
    Console.WriteLine("DoSthAsync: START");
    await Task.Delay(100);
    Console.WriteLine("DoSthAsync: END");
}

但是通过如下代码就可以完成运行。这将与同步上下文有关，也就是SynchronizationContext类

static async Task Main(string[] args)
{
    Console.WriteLine("Test:Start");
    DoSthAsync().GetAwaiter().GetResult();
    Console.WriteLine("Test:END");
    Console.ReadLine();
}
static async Task DoSthAsync()
{
    Console.WriteLine("DoSthAsync: START");
    await Task.Delay(100).ConfigureAwait(continueOnCapturedContext:false);
    Console.WriteLine("DoSthAsync: END");
}

SynchronizationContext与Task回调处理

Task不仅可以让TaskScheduler调用，还可以使用SynchronizationContext。
TaskAwaiter在实际注册回调的时候会对回调进行包装，决定了回调会在哪里执行。
当SynchronizationContext.Current不为为null时将会到SynchronizationContext上执行回调，为空是才会去判断TaskScheduler。（具体示意图如下）

自定义SynchronizationContext

class MaxConcurrencySynchronizationContext: SynchronizationContext
{
    /// <summary>
    /// 创建一个信号量
    /// </summary>
    private readonly SemaphoreSlim _semaphore;
    public MaxConcurrencySynchronizationContext(int maxConcurrencyLevel) =>
        _semaphore = new SemaphoreSlim(maxConcurrencyLevel);
    public override void Post(SendOrPostCallback d, object state)
    {
        _semaphore.Wait();
        try
        {
            Console.WriteLine("MaxConcurrencySynchronizationContext.ThreadId:{0}",Thread.CurrentThread.ManagedThreadId);
            d(state);
        }
        finally
        {
            _semaphore.Release();
        }
    }
    public override void Send(SendOrPostCallback d, object state) => throw new NotImplementedException();
}

添加测试代码

static void Main(string[] args)
{
    Task.Run(() =>
    {
        var mcs = new MaxConcurrencySynchronizationContext(1);
        // 设置为我们自定义的同步上下文
        SynchronizationContext.SetSynchronizationContext(mcs);
        mcs.Post(_ => { Test(); }, null);
    });
    Console.ReadLine();
}
static void Test()
{
    Console.WriteLine("Test:Start");
    DoSthAsync().GetAwaiter().GetResult();
    Console.WriteLine("Test:END");
}
static async Task DoSthAsync()
{
    Console.WriteLine("DoSthAsync: START");
    await Task.Delay(100);
    Console.WriteLine("DoSthAsync: END");
}

由于最大等待数量为1，所以当我们第二次在Task.Delay的时候就会被阻塞起。

而之前我们看到的.ConfigureAwait(continueOnCapturedContext:false);,这段代码的意思就是：不管你有没有同步上下文我都会到TaskScheduler上去执行。

static void Main(string[] args)
{
    Task.Run(() =>
    {
        var mcs = new MaxConcurrencySynchronizationContext(1);
        // 设置为我们自定义的同步上下文
        SynchronizationContext.SetSynchronizationContext(mcs);
        mcs.Post(_ => { Test(); }, null);
    });
    Console.ReadLine();
}
static void Test()
{
    Console.WriteLine("Test:Start");
    DoSthAsync().GetAwaiter().GetResult();
    Console.WriteLine("Test:END");
}
static async Task DoSthAsync()
{
    Console.WriteLine("DoSthAsync: START");
    await Task.Delay(100).ConfigureAwait(false);
    Console.WriteLine("DoSthAsync: END");
}

ValueTask

ValueTask与Task类似，但它是值类型的。ValueTask<TResult>提供了一个AsTask方法，可根据需要获取的常规task（例如用作Task.WhenAll或Task.WhenAny调用的某个元素），不过多数情况下只是安装普通task那样调用await操作。

ValueTask<TResult>相比于Task<TResult>优势何在？其实体现在堆内存分配和垃圾回收上。Task<TResult>是一个类，虽然有时异步基础架构会复用已创建的Task<TResult>对象。但多数async方法需要创建新的Task<TResult>对象。一般情况下，.NET创建对象的性能消耗不足为虑，但如果频繁创建对象或者遇到性能敏感的代码，就需要尽量避免创建新对象。

如果在async方法中对某个尚未完成的操作使用await,那么创建对象是不可避免的，虽然此时方法会立即返回，但它需要安排一个延续。当操作完成时，该延续负责执行async方法中的其他语句。大部分async方法中的操作不会在await前执行完成。此时使用使用ValueTask<TResult>没有任何优势，可能还会导致性能下降。

有时task在await之前便已完成，这个时候就能体现出ValueTask<TResult>的强大了

public sealed class ByteStream : IDisposable
{
    private readonly Stream stream;
    private readonly byte[] buffer;
    private int position;
    private int bufferedBytes;
    public ByteStream(Stream stream)
    {
        this.stream = stream;
        // 8KB 缓冲区的大小，根本不需要await操作
        buffer = new byte[1024 * 8];
    }
    public async ValueTask<byte?> ReadByteAsync()
    {
        if (position == bufferedBytes)
        {
            position = 0;
            // 读取数据且不需要同步上下文执行
            bufferedBytes = await stream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false);
            // 读到末尾了
            if (bufferedBytes == 0)
            {
                return null;
            }
        }
        // 返回缓冲区的下一个字节
        return buffer[position++];
    }
    public void Dispose()
    {
        stream.Dispose();
    }
}