在不阻塞线程的情况下异步检查值？

Question

我的应用程序中有一个模式，我想发送一些命令或启动一些 IO 工作并等待它完成，或者有一些机制来知道它是否已经完成。

为此，我计划使用async/await 模式。我假设大多数人在编码时都达到了这种模式。

while(something)
    DoNothing();

DoNothing() 通常最终会占用一些 CPU 时间或完全停止程序。我认为解决这个问题的正确方法是使用async/await 模式。

就我而言，我想出了以下简单的方法

public override async Task<Boolean> PerformProcessingAsync()
{
    StartSomeIOProcessing();
    while (TheIOProcessingResult == null)
        await Task.Yield();
    return true;
}

它开始一些 IO 处理，然后等待结果被实例化。与此同时，尽管它调用Task.Yield 返回到调用上下文，然后可以继续工作，并将此方法的延续（下一个while 循环迭代）放到调用堆栈上。

这是正确的解释吗，这是解决上述情况的正确方法吗？

编辑：在我面临的更具体的情况下......我还维护另一个执行 IO 工作的库，主要是读取和写入 SerialPort 对象。这个库的工作原理是拥有一个ConcurrentQueue 的读取或写入，它针对特定端口处理它。一个Task“位于此队列的末尾”并在执行过程中消耗工作。

大多数阅读工作只是查询一些值，对其进行解析，然后触发一个事件NewData(double myNewData)，该事件由 UI 监听。在 UI 中，状态表示保存了 SerialPorts 数据。在状态表示中，NewData 事件被处理，该事件更新它对应的值。

写入以相同的方式完成，但完成时不会触发事件，只是将端口写入。判断它是否成功的唯一方法是等待读取更新状态。 （遗憾的是，由于硬件的行为方式，没有更好的方法可以做到这一点。

---

我当前的应用程序利用这个库来执行它的 IO 工作。读取会定期发送到库，以使 UI 保持来自端口的新值...当用户单击按钮时，写入会发送到库以从硬件发出命令。

我希望确保写入是以编程方式发生的。我能想到的唯一方法是将写入发送到库，然后等待读取更新写入效果的数据。

因此循环。

Answer 1

我现在想确保写入是以编程方式发生的。我能想到的唯一方法是将写入发送到库，然后等待读取更新写入效果的数据。

是的，这可能是唯一的方法，但还有比旋转等待查看读取更新是否发生更好的方法。

这是处理处理的示例方法，就像我们在 the chat 中谈到的那样。总之，您有一个IDataRequest 队列，在这些请求中，他们持有一个TaskCompletionSource<T>，表示发送数据完成。一旦您向设备发出请求并获得响应，您就可以设置完成源的结果。我将它与您现有的基于事件的实现结合起来，但老实说，我会放弃该事件，让调用者在等待RequestTemp() 的结果后更新 UI。

public interface IDataRequest
{
    bool TrySetException(Exception ex);
}

public abstract class DataRequest<T> : IDataRequest
{
    public TaskCompletionSource<T> RequestTask { get; } = new TaskCompletionSource<T>();

    public bool TrySetException(Exception ex)
    {
        return RequestTask.TrySetException(ex);
    }
}

public class TempRequest : DataRequest<double>
{
}

public class RpmRequest : DataRequest<int>
{
}

public sealed class DeviceManager : IDisposable
{

    private readonly Task _workerThread;
    private readonly BlockingCollection<IDataRequest> _queue;
    private readonly SerialPort _serialPort;

    public DeviceManager()
    {
        _queue = new BlockingCollection<IDataRequest>();
        _workerThread = Task.Factory.StartNew(ProcessQueue, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        _serialPort = //...
    }
    
    public event EventHandler<double> TempUpdate;
    public event EventHandler<int> RpmUpdate;

    public Task<double> RequestTemp()
    {
        var request = new TempRequest();
        _queue.Add(request);
        return request.RequestTask.Task;
    }

    public Task<int> RequestRpm()
    {
        var request = new RpmRequest();
        _queue.Add(request);
        return request.RequestTask.Task;
    }
    public void Dispose()
    {
        _queue.CompleteAdding();
        _workerThread.Wait();
    }

    private void ProcessQueue()
    {
        foreach (var dataRequest in _queue.GetConsumingEnumerable())
        {
            try
            {
                if (dataRequest is TempRequest)
                {
                    DoTempRequest((TempRequest)dataRequest);
                }
                else if (dataRequest is RpmRequest)
                {
                    DoRpmRequest((RpmRequest)dataRequest);
                }
                else
                {
                    throw new NotSupportedException($"A Request of type {dataRequest.GetType()} is not supported.");
                }
            }
            catch (Exception ex)
            {
                dataRequest.TrySetException(ex);
            }
        }
    }

    private void DoTempRequest(TempRequest dataRequest)
    {
        _serialPort.WriteLine("Temp ?");
        var line = _serialPort.ReadLine();
        double result;

        //I am deliberately using Parse instead of TryParse so responses that 
        //fail to parse will throw and get their exception propagated up via the 
        //catch in ProcessQueue().
        result = double.Parse(line);
        
        //Sends the info back to the caller saying it is done and what the result was.
        dataRequest.RequestTask.TrySetResult(result);

        //Raises the event so subscribers know the new value.
        OnTempUpdate(result);
    }

    private void DoRpmRequest(RpmRequest dataRequest)
    {
        _serialPort.WriteLine("RPM ?");
        var line = _serialPort.ReadLine();
        int result;
        result = int.Parse(line);
        
        dataRequest.RequestTask.TrySetResult(result);
        OnRpmUpdate(result);
        
    }

    private void OnTempUpdate(double result)
    {
        TempUpdate?.Invoke(this, result);
    }

    private void OnRpmUpdate(int result)
    {
        RpmUpdate?.Invoke(this, result);
    }
}

Answer 2

我会将StartSomethingIOProcessing 提取到返回结果的单独Task 中。 Await ，然后检查结果。

public async PerformProcessingAsync()
{
    var ioProcessingResult = await StartSomeIOProcessing();
    return (null != ioProcessingResult);
}

private Task<TheIOProcessingResultType> StartSomeIOProcessing()
{
    return Task.Run(()=>
    {
        return StartSomeIOProcessing();
    });
}

Answer 3

我相信您误解了异步/等待操作。 await 关键字实际上用于等待异步操作完成，因此您无需再实现 while 循环并检查它是否为空。

public override async Task<Boolean> PerformProcessingAsync()
{
    await Task.Run((() =>
    {
            StartSomeIOProcessing();
    });
    return true;
}