c# - 在 C# 中正确使用取消 token

Question

我最近接触了 C# 语言，并致力于从 cassandra 中获取数据，所以我正在使用下面的代码从 Cassandra 获取数据，并且工作正常。

我唯一的问题是我的 ProcessCassQuery方法-我路过CancellationToken.None到我的 requestExecuter可能不正确的功能。处理这种情况的正确方法应该是什么，我应该怎么做才能正确处理它？

/**
 *
 * Below method does multiple async calls on each table for their corresponding id's by limiting it down using Semaphore.
 *
 */
private async Task<List<T>> ProcessCassQueries<T>(IList<int> ids, Func<CancellationToken, int, Task<T>> mapperFunc, string msg) where T : class
{
    var tasks = ids.Select(async id => 
    {
        await semaphore.WaitAsync();
        try
        {
            ProcessCassQuery(ct => mapperFunc(ct, id), msg);
        }
        finally
        {
            semaphore.Release();
        }
    });

  return (await Task.WhenAll(tasks)).Where(e => e != null).ToList();
}

// this might not be good idea to do it. how can I improve below method?
private Task<T> ProcessCassQuery<T>(Func<CancellationToken, Task<T>> requestExecuter, string msg) where T : class
{
    return requestExecuter(CancellationToken.None);
}

Answer 1

如 official documentation 中所述，取消 token 允许传播取消信号。例如，这对于取消由于某种原因不再有意义或只是花费太长时间的长时间运行的操作非常有用。
CancelationTokenSource将允许您获得可以传递给 requestExecutor 的自定义 token .它还将提供取消正在运行的 Task 的方法。 .

private CancellationTokenSource cts = new CancellationTokenSource();

// ...

private Task<T> ProcessCassQuery<T>(Func<CancellationToken, Task<T>> requestExecuter, string msg) where T : class
{
    return requestExecuter(cts.Token);
}

示例

让我们看一个不同的最小/虚拟示例，以便我们可以查看它的内部。

考虑以下方法，GetSomethingAsync这将产生每秒返回一个递增的整数。

调用 token.ThrowIfCancellationRequested将确保 TaskCanceledException如果此进程被外部操作取消，则抛出。还可以采取其他方法，例如检查是否token.IsCancellationRequested是真的，并为此做点什么。

private static async IAsyncEnumerable<int> GetSomethingAsync(CancellationToken token)
{
    Console.WriteLine("starting to get something");

    token.ThrowIfCancellationRequested();

    for (var i = 0; i < 100; i++)
    {
        await Task.Delay(1000, token);
        yield return i;
    }

    Console.WriteLine("finished getting something");
}

现在让我们构建 main 方法来调用上述方法。

public static async Task Main()
{    
    var cts = new CancellationTokenSource();

    // cancel it after 3 seconds, just for demo purposes
    cts.CancelAfter(3000);
    // or: Task.Delay(3000).ContinueWith(_ => { cts.Cancel(); });

    await foreach (var i in GetSomethingAsync(cts.Token))
    {
        Console.WriteLine(i);
    }
}

如果我们运行它，我们将得到如下所示的输出:

starting to get something
0
1
Unhandled exception. System.Threading.Tasks.TaskCanceledException: A task was canceled.

当然，这只是一个虚拟示例，取消可以由用户操作或某些发生的事件触发，它不一定是计时器。