长时间运行同步实现返回Task的接口

Question

我正在使用这个问题作为我的问题的基础.

---

TL; DR:如果您不打算在异步包装器中包装同步代码,那么如何处理长期运行的线程阻塞方法,这些方法实现了一个需要异步实现的接口方法？

---

假设我有一个连续运行的应用程序来处理工作队列.它是一个服务器端应用程序(主要是无人值守运行),但它有一个UI客户端,可以根据业务流程的要求对应用程序的行为进行更细粒度的控制:在执行期间启动,停止,调整参数,获得进度等

有一个业务逻辑层,服务作为依赖项注入其中.
BLL为这些服务定义了一组接口.

我想让客户端保持响应:允许UI客户端与正在运行的进程交互,我还希望有效地使用线程,因为进程需要可伸缩:根据工作可能有任意数量的异步数据库或磁盘操作在队列中.因此,我正在使用async/await "一直".

为此,我在服务接口中有一些方法,显然是为了鼓励async/await和支持取消,因为它们采用a CancellationToken,以"Async"命名,并返回Tasks.

我有一个数据存储库服务,执行CRUD操作以持久保存我的域实体.让我们说,目前,我正在使用一个本身不支持异步的API.在将来,我可以用一个替换它,但目前数据存储库服务同步执行大部分操作,其中许多是长时间运行的操作(因为数据库IO上的API阻塞).

现在,我知道返回Tasks的方法可以同步运行.我的服务类中实现BLL中接口的方法将按照我的解释同步运行,但是消费者(我的BLL,客户端等)将假设它们是1:异步运行或2:同步运行很短的时间.方法不应该做的是在异步调用中包装同步代码Task.Run.

我知道我可以在界面中定义同步和异步方法.
在这种情况下,我不想这样做,因为我试图使用异步"一路"语义,因为我不是在编写一个API供客户使用; 如上所述,我不想稍后将BLL代码从使用同步版本更改为使用异步版本.

这是数据服务接口:

public interface IDataRepository
{
    Task<IReadOnlyCollection<Widget>> 
        GetAllWidgetsAsync(CancellationToken cancellationToken);
}

它的实施:

public sealed class DataRepository : IDataRepository
{
    public Task<IReadOnlyCollection<Widget>> GetAllWidgetsAsync(
        CancellationToken cancellationToken)
    {
        /******* The idea is that this will 
        /******* all be replaced hopefully soon by an ORM tool. */

        var ret = new List<Widget>();

        // use synchronous API to load records from DB
        var ds = Api.GetSqlServerDataSet(
            "SELECT ID, Name, Description FROM Widgets", DataResources.ConnectionString);

        foreach (DataRow row in ds.Tables[0].Rows)
        {
            cancellationToken.ThrowIfCancellationRequested();
            // build a widget for the row, add to return.  
        }

        // simulate long-running CPU-bound operation.
        DateTime start = DateTime.Now;
        while (DateTime.Now.Subtract(start).TotalSeconds < 10) { }

        return Task.FromResult((IReadOnlyCollection<Widget>) ret.AsReadOnly());
    }
}

BLL:

public sealed class WorkRunner
{
    private readonly IDataRepository _dataRepository;
    public WorkRunner(IDataRepository dataRepository) => _dataRepository = dataRepository;

    public async Task RunAsync(CancellationToken cancellationToken)
    {
        var allWidgets = await _dataRepository
            .GetAllWidgetsAsync(cancellationToken).ConfigureAwait(false);

        // I'm using Task.Run here because I want this on 
        // another thread even if the above runs synchronously.
        await Task.Run(async () =>
        {
            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();
                foreach (var widget in allWidgets) { /* do something */ }
                await Task.Delay(2000, cancellationToken); // wait some arbitrary time.
            }
        }).ConfigureAwait(false);
    }
}

演示和演示逻辑:

private async void HandleStartStopButtonClick(object sender, EventArgs e)
{
    if (!_isRunning)
    {
        await DoStart();
    }
    else
    {
        DoStop();
    }
}

private async Task DoStart()
{
    _isRunning = true;          
    var runner = new WorkRunner(_dependencyContainer.Resolve<IDataRepository>());
    _cancellationTokenSource = new CancellationTokenSource();

    try
    {
        _startStopButton.Text = "Stop";
        _resultsTextBox.Clear();
        await runner.RunAsync(_cancellationTokenSource.Token);
        // set results info in UI (invoking on UI thread).
    }
    catch (OperationCanceledException)
    {
        _resultsTextBox.Text = "Canceled early.";
    }
    catch (Exception ex)
    {
        _resultsTextBox.Text = ex.ToString();
    }
    finally
    {
        _startStopButton.Text = "Start";
    }
}

private void DoStop()
{
    _cancellationTokenSource.Cancel();
    _isRunning = false;
}

所以问题是:你如何处理长期运行的阻塞方法,这些方法实现了一个需要异步实现的接口方法？这是一个例子,它最好打破"同步代码的无异步包装"规则吗？

Answer 1

您没有公开同步方法的异步包装器.您不是外部库的作者,您是客户.作为客户端,您正在将库API 调整为服务接口.

针对同步方法使用异步包装器的建议的主要原因是(从问题中引用的MSDN文章中总结):

1. 确保客户端了解任何同步库函数的真实性质
2. 让客户端控制如何调用该函数(异步或同步).
3. 通过每个函数的2个版本来避免增加库的表面积

对于您的服务接口,通过仅定义异步方法,您选择异步调用库操作,无论如何.你实际上在说,无论(1)如何,我都选择了(2).并且您给出了合理的理由 - 从长远来看,您知道您的同步库API将被替换.

作为一个侧面点,即使您的外部库API函数是同步的,它们也不会长时间运行CPU绑定.正如你所说,他们阻止IO.它们实际上是IO绑定的.它们只是阻塞线程等待IO而不是释放它.