4

我正在开发基于 xamarin 和 .net 5 async/await 的 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 个 pro/con 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 上下文中的任何 UI（WinForm、WPF 等）OnCreate 方法来调用 RunAsync

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 等待调用之后我什么都做不了，因为它永远不会从 Task.WhenAny(...) 返回。我需要在那里执行状态检查，但我需要启动这个 pro/cons 方法，因为我的检查等待 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();
}

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

4

1 回答 1

2

好的，经过大量阅读（没有找到）和@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]");
}

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

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

由于不等待此调用，因此在调用完成之前继续执行当前方法。考虑将“等待”运算符应用于调用结果。

可以通过这种方式轻松抑制

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

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

于 2013-08-12T07:18:26.490 回答