如何在复杂的异步等待场景中实现超时和取消[重复]

Question

我很困惑如何为以下代码实现取消/重试机制，为了简洁起见，它主要是伪代码：

class CLI {
    async void Main() {
        var response = await bot.GetSomething(); // final usage
    }
}

class Bot {
    private void StartTask() {
      _taskCompSource = new TaskCompletionSource<object>();
    }

    private async Task<T> ResultPromise<T>() {
      return await _taskCompSource.Task.ContinueWith(t => t.Result != null ? (T)t.Result : default(T));
    }

    // send request
    public Task<string> GetSomething() {
        StartTask();
        QueueRequest?.Invoke(this, new Request(...)); // complex non-sync request, fired off here
        return ResultPromise<string>();
    }

    // receive response
    public void Process(Reply reply) {
        // process reply
        _taskCompSource.SetResult("success");
    }
}

class BotManager {
    // bot.QueueRequest += QueueRequest;
    // _service.ReplyReceived += ReplyReceived;

    private void QueueRequest(object sender, Request request) {
      _service.QueueRequestForSending(request);
    }

    private async void ReplyReceived(object sender, Reply reply) {
      GetBot().Process(reply);
    }
}

class Service {
    private Dictionary<string, Queue<Request>> _queues;

    // send receive loop
    private async void HttpCallback(IAsyncResult result) {
       while (true) {
        // if reply received then call ReplyReceived?.Invoke(this, reply);
        // check if request already in progress // ##ISSUE IS HERE##
        // if not send another request
        // keep connection to server open by feeding spaces
       }
    }

    public void QueueRequestForSending(Request request) {
      GetQueueForBot().Enqueue(request);
    }
}

我想在await bot.GetSomething(); 上实现超时，但不确定如何处理这种断开连接的性质。我试过了：

static class Extensions {
    private static async Task<T> RunTaskWithRetry<T>(Func<Task<T>> taskFunc, int retries, int timeout) {
      do {
        var task = taskFunc(); // just adds another request to queue which never gets ran because the previous one is waited forever to return, this needs to cancel the previous request, from here, but how?

        await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false);
        if (task.Status == TaskStatus.RanToCompletion) {
          return task.Result;
        }

        retries--;
      }
      while (retries > 0);

      return default(T);
    }

    public async static Task<T> WithRetry<T>(this Task<T> task, int retries = 3, int timeout = 10000) {
      return await RunTaskWithRetry(async () => await task, retries, timeout);
    }
}

可以像await bot.GetSomething().WithRetry(); 一样调用它，但这里的问题是它只是将另一个请求添加到队列中，而不是取消或删除现有的请求。要取消，我只需从awaiting 列表中删除现有请求。问题是我不知道如何从扩展方法的位置一路做到这一点。

我想知道我可以用来实现超时和取消的任何可能机制。

Answer 1

从扩展方法到哪里，一直不知道怎么弄。

控制TaskCompletionSource<T> 的代码也必须控制它的取消。该代码可以使用TaskCompletionSource<T>.TrySetCanceled 来实现。如果您还需要取消StartTask 操作，那么您可能希望对using CancellationToken.Register 建模。我有一个类似的async wait queue，除了它管理一个TaskCompletionSource<T> 实例队列，而不仅仅是一个。

顺便说一句，重试逻辑是不确定的。 WhenAny 的结果是完成的任务，所以检查Status 是没有必要的；并且Result 应替换为await 以避免AggregateException 包装器：

var completed = await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false);
if (task == completed) {
  return await task;
}

Answer 2

你可以使用 Task 代替 await 这个提议

  Task responseTask = bot.GetSomething(); // final usage
  responseTask.Wait(5000); //Timeout in 5 seconds.
  var result = responseTask.Result;

如果您想使用更复杂的系统，例如取消政策，请阅读这篇可能对您有所帮助的帖子：https://johnthiriet.com/cancel-asynchronous-operation-in-csharp/