.NET中的异步编程：最佳实践

计划

• 用异步/等待解决的方法存在问题。
• 有争议的设计示例。
• 现实生活中的一项任务，我们将异步解决。

异步/等待和有待解决的问题

异步/等待路径

打回来

Void Foo(params, Action callback) {…}
 

Void OurMethod() {
    …//synchronous code
 
    Foo(params,() =>{
       …//asynchronous code;continuation
    });
}

更多回调

void Foo(params, Action callback) {...} 
void Bar(Action callback) {...}
void Baz(Action callback) {...}

void OurMethod() {
    ... //synchronous code
    
    Foo(params, () => { 
      ... //continuation 1 
      Bar(() => {
        //continuation 2
        Baz(() => {
          //continuation 3
        }); 
      });
    });
}

回调：例外

void Foo(params, Action onSuccess, Action onFailure) {...}


void OurMethod() {
    ... //synchronous code 
    Foo(params, () => {
      ... //asynchronous code on success 
    },
    () => {
        ... //asynchronous code on failure
    }); 
}

合并延续

Task ourMethod() {
  return Task.RunSynchronously(() =>{
    ... //synchronous code
  })
  .ContinueWith(_ =>{
    Foo(); //continuation 1
  })
  .ContinueWith(_ =>{
    Bar(); //continuation 2
  })
  .ContinueWith(_ =>{
    Baz(); //continuation 3
  })
}

开始和继续任务

Task.Factory.StartNew(Action, 
  TaskCreationOptions, 
  TaskScheduler, 
  CancellationToken
)
Task.ContinueWith(Action<Task>, 
  TaskContinuationOptions, 
  TaskScheduler, 
  CancellationToken
)

TaskCompletionSource

var  tcs  =  new   TaskCompletionSource<int>(
       TaskContinuationsOptions.RunContinuationsAsynchronously  
) ;
lock(mylock)
{  
    tcs.SetResult(O); 
});

亲子层次

Task.Factory.StartNew(() => 
{
  //... some parent activity

   Task.Factory.StartNew(() => {
      //... some child activity
   })

})
.ContinueWith(...) // don’t wait for child

Task.Factory.StartNew(() => 
{
  //... some parent activity
  Foo(); 

})
.ContinueWith(...) // still wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... parent task to attach is in ThreadStatic
   }, TaskCreationOptions.AttachedToParent); 
}

Task.Factory.StartNew(() => 
{
  //... some parent activity

  Foo();
}, TaskCreationOptions.DenyChildAttach)
.ContinueWith(...) // don’t wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... some child activity
   }, TaskCreationOptions.AttachedToParent); 
}

Task.Run(() => 
{
  //... some parent activity

  Foo(); 

})
.ContinueWith(...) //don’t wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... some child activity
    }, TaskCreationOptions.AttachedToParent); 
}

• 常规继续将成功完成；将发生RunToCompletion；
• 任务将是错误的；
• 取消将发生；
• 这项任务根本不会完成，而是处于某种困境。
• 选项-“取决于”。

消除

Task.Factory.StartNew(() => 
{
    throw new OperationCanceledException(); 
});

                                                      Failed

Task.Factory.StartNew(() => 
{
    throw new OperationCanceledException(cancellationToken); 
}, cancellationToken);

                                                      Canceled

Task.Factory.StartNew(() => 
{
    cancellationToken.ThrowIfCancellationRequested(); 
}, cancellationToken);
                                                       Canceled

深栈

Task.Factory.StartNew(() => 
{
    Foo();
}, cancellationToken);

  void Foo() { 
     Bar() {
       ...
          Baz() {
             //how to get cancellation token?
          } 
    }
}

static AsyncLocal<Cancelation> asyncLocalCancellation;

Task.Factory.StartNew(() => 
{
     asyncLocalCancellation.Set(cancellationToken) 
    Foo();
}, cancellationToken); // use AsyncLocal to put cancellation int

  void Foo() { 
     async Bar() {
      ...
         Baz() {
             asyncLocalCancellation.Value.CheckForInterrupt(); 
         }
   } 
}

认知复杂性

Task.Factory.StartNew(Action, 
    TaskCreationOptions, 
    TaskScheduler, 
    CancellationToken
)
                                                   JetBrains.Lifetimes

lifetime.Start(TaskScheduler, Action) //puts lifetime in AsyncLocal

lifetime.StartMainRead(Action) 
lifetime.StartMainWrite(TaskScheduler, Action) 
lifetime.StartBackgroundRead(TaskScheduler, Action)

async Task MyFuncAsync() { 
  synchronousCode();
   await InnerAsync();
   await Task.Yield(); //guaranteed !IsCompleted 
   continuationCode();
}

async Task MyFuncAsync() { 
  synchronousCode();

    await InnerAsync().ConfigureAwait(false);
    continuationCode(); 
}

async Task MyFuncAsync() { 
  synchronousCode();

    await InnerAsync().ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode(); //code must be absolutely context-agnostic
}

死锁

async Task MyFuncAsync() { //UI thread 
  synchronousCode();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode();
}
myFuncAsync().Wait() //on UI thread

async Task OnBluttionClick() { //UI thread 
  int v = Button.Text.ParseInt();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
  Button.Text.Set((v+1).ToString());
}
myFuncAsync().Wait() //on UI thread

async Task MyFuncAsync() { //UI thread 
  synchronousCode();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode(); //The same UI context
}
PumpUntil(() => task.IsCompleted);
//VS synchronization contexts always pump on any Wait

改变背景

async Task MyFuncAsync() { 
  synchronousCode(); // on initial context

    await myTaskScheduler;
    continuationCode(); //on scheduler context 
}

结构化并发

async Task MyFuncAsync() { 
  synchronousCode(); // on initial context

    await Task.Factory.StartNew(() => {...}, myTaskScheduler);
    continuationCode(); //on initial context 
}

顺序行为

async Task MyAsync() {

  var task1 = StartTask1Async();
  await task1;

  var task2 = StartTask2Async();
  await task2; 
}

并发行为

async Task MyAsync() {
  var task1 = StartTask1Async();
  var task2 = StartTask2Async();

  await task1;
  await task2; 
}

异常处理

async Task MyAsync() {
  var task1 = StartTask1Async();
  var task2 = StartTask2Async(); 

  await task1;
  await task2;

  // if task1 throws exception and task2 throws exception we only throw and
  // handle task1’s exception

  //4.0 -> 4.5 framework: unhandled exceptions now don’t crush process
  //still visible in UnobservedExceptionHandler
}

async  Task  MyAsync(CancellationToken cancellationToken)  {  

  await  SomeTask1  Async(cancellationToken); 
 
  await  Some Task2Async( cancellation  Token); 
  //you should always pass use async API with cancelationToken  if possible 
} 
  
try { 
    await  MyAsync( cancellation  Token); 
} catch (OperationException e) { // do nothing: OCE happened
} catch (Exception e) { 
    log.Error(e);
}

在实践中

同步方式

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
void SynchronousWorker(...) {
  File f = blockingQueue.Dequeue(); 
  ProcessedFile p = ProcessInParallel(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

异步方式

DataTable<File, ProcessedFile> sharedMemory;
// in any thread
async Task WorkerAsync(...) {

  File f = blockingQueue.Dequeue(); 

  ProcessedFile p = ProcessInParallel(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

简化并行代码

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

骨架解决方案

Dictionary<Action<ProcessedFile>, Action<ProcessedFile>[]> ExecuteBefore; async Task<ProcessedFile> ProcessInParallelAsync() {
  var res = new ProcessedFile();


  // lots of work with toposort, locks, etc.

  return res; 
}

异步懒

Dictionary<Action<ProcessedFile>, Action<ProcessedFile>[]> ExecuteBefore;
async Task<ProcessedFile> ProcessInParallelAsync() {
  var res = new ProcessedFile();
  var lazy = new Dictionary<Action<ProcessedFile>, Lazy<Task>>(); 
  foreach ((action, beforeList) in ExecuteBefore)
    lazy[action] = new Lazy<Task>(async () => 
    {
      await Task.WhenAll(beforeList.Select(b => lazy[b].Value)) 
      await Task.Yield();
      action(res);
}
  await Task.WhenAll(lazy.Values.Select(l => l.Value)) 
  return res;
}

摆脱锁

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);

    lock (_lock) {
      sharedMemory.add(f, p);
   }
 }

Channel<Pair<File, ProcessedFile>> output;
// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);
  
  await output.WriteAsync(); 
}

摘要

• 避免异步代码中的同步。
• 串行代码比并行代码简单。
• 异步代码可以很简单，并使用最少的参数和更改其行为的隐式上下文。

读什么

• TAP文件
• 异步指导
• 一生
• medium.com/@elizarov/deadlocks-in-non-hierarchical-csp-e5910d137cc

-10 . DotNext .