c# - 使用 async/await 运行多个无限循环

Question

我正在开发基于 xamarin 和 .net 5 async/awaits 的 android messanger 应用程序。

在我的应用程序中，我有生产者/消费者模式来处理无限循环产生的消息。

例如 ReadTcpClientAsync 生产者:

async Task ReadTcpClientAsync(CancellationToken cancellationToken)
{
    cde.Signal();
    while (!cancellationToken.IsCancellationRequested)
    {
        byte[] buffer = await atc.ReadAsync(cancellationToken);
        // queue message...
    }
}

或发送消息并等待 WriteAsync 的 SendStatementsAsync 消费者

private async Task SendStatementsAsync(CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        var nextItem = await _outputStatements.Take();
        cancellationToken.ThrowIfCancellationRequested();
        // misc ...
        await atc.WriteAsync(call.Serialize());
    }
}

有些消费者只是等待接听电话

 var update = await _inputUpdateStatements.Take();

这种结构在测试中效果很好，但我认为有一种方法犯了一个大错误。此方法旨在运行整个客户端后端，同时启动 3 个赞成/反对 while (true) 循环。

这里是:

public async Task RunAsync()
{
   _isRunning = true;
   _progress.ProgressChanged += progress_ProgressChanged;
    await InitMTProto(_scheme).ConfigureAwait(false); // init smth...
    // various init stuf...     
    await atc.ConnectAsync().ConfigureAwait(false); // open connection async
    // IS IT WRONG?
    try
    {                                   
        await Task.WhenAny(SendStatementsAsync(_cts.Token),
                               ReadTcpClientAsync(_cts.Token),
                               ProcessUpdateAsync(_cts.Token, _progress)).ConfigureAwait(false);
    }
    catch (OperationCanceledException oce)
    {   

    }
    catch (Exception ex)
    {

    }
}

暂时忘掉 android，考虑在 UI 上下文中调用 RunAsync 的任何 UI(WinForm、WPF 等)OnCreate 方法

protected async override void OnCreate(Bundle bundle)
{
    // start RA 
    await client.RunAsync()
    // never gets here - BAD, but nonblock UI thread - good
    Debug.WriteLine("nevar");
}

所以，如您所见有问题。在调用 RunAsync await 之后我什么也做不了，因为它永远不会从 Task.WhenAny(...) 返回。我需要在那里执行状态检查，但我需要启动此优缺点方法，因为我的检查等待 ManualResetEvent:

if (!cde.Wait(15000))
{
    throw new TimeoutException("Init too long");
}

此外，我的支票也是异步的，它的工作原理非常棒:)

public async Task<TLCombinatorInstance> PerformRpcCall(string combinatorName, params object[] pars)
{
    // wait for init on cde ...
    // prepare call ...

    // Produce
    ProduceOutput(call);

    // wait for answer
    return await _inputRpcAnswersStatements.Take();
}

我想我应该使用另一种方法来启动这个无限循环，但我已经有了异步任务方法 - 所以我真的不知道该怎么做。有什么帮助吗？

Answer 1

好的，经过大量阅读(未找到任何内容)和@svick 的建议后，我决定调用此方法而不“等待”作为单独的 Task.Run。Aso 我决定在 ThreadPool 中运行它。

我的最终代码是:

try
{                                   
    /*await Task.WhenAny(SendStatementsAsync(_cts.Token), 
           ReadTcpClientAsync(_cts.Token),
           ProcessUpdateAsync(_cts.Token, _progress)).ConfigureAwait(false);*/
    Task.Run(() => SendStatementsAsync(_cts.Token)).ConfigureAwait(false);
    Task.Run(() => ReadTcpClientAsync(_cts.Token)).ConfigureAwait(false);
    Task.Run(() => ProcessUpdateAsync(_cts.Token, _progress)).ConfigureAwait(false);
    Trace.WriteLineIf(clientSwitch.TraceInfo, "Worker threads started", "[Client.RunAsync]");
}

一切都如预期的那样正常..我不确定它会在异常处理中引起什么问题，因为我知道它们会丢失

当然这样的调用会产生警告

Because this call is not awaited, execution of the current method continues before the call is completed. Consider applying the 'await' operator to the result of the call.

这样可以很容易地抑制

// just save task into variable
var send = Task.Run(() => SendStatementsAsync(_cts.Token)).ConfigureAwait(false); 

此外，如果有人知道更好的解决方案，我将不胜感激。