c# - 排队异步操作，让它们在 MVVM 应用程序中一次运行一个

Question

我目前正在构建一个 MVVM 应用程序，我的一个 View 模型使用通过依赖注入(inject)注册的服务。此服务针对各种第 3 方应用程序运行 powershell cmdlet 或 http REST 命令，这些应用程序在同时收到多个请求时不太满意。

这就是为什么我希望能够从 UI 触发多个操作(不阻塞它)，但要确保服务一次只处理一个。我的 UI 元素将同时显示它们是否正在工作或等待。

我尝试实现 TPL ActionBlock，但到目前为止，我的所有操作都同时运行，这是我发现让它们在队列中工作的唯一方法会阻塞 UI，直到所有任务完成。

这是我所做的:

我的 View 模型包含一个 ObservableCollection 元素，其中包含两个列表(一个嵌套在另一个列表中)在 UI 上，它看起来像一个项目列表，可以展开以显示一个小 TreeView 。

我想要的是，每次我展开一个项目时， TreeView 中的所有子项目都会通过服务检查它们在 3rd 方应用程序中的状态。
UI子项中的方法如下所示:

private async Task<bool> UpdateSubItemsStatus()
    {
        foreach (var item in connectorsMenuItems)
        {
            await parent.Library.EnqueueConnectorOperations(Connectors.service.OperationType.CheckPresence, parent.CI, AssetId, item.ConnectorID, parent.ConnectorInterfaces.Single(c => c.ItemId == AssetId).ItemsConnectorPresence.Single(i => i.ConnectorId == item.ConnectorID));
        }
        return true;
    }

在这里，“parent”是第一级项目，“parent.Library”是托管所有内容的主视图模型。

在 View 模型上，检索 this 的方法如下:

public async Task EnqueueConnectorOperations(OperationType operationType, ConfigurationItem ci, Guid itemId, Guid ConnectorID, ItemConnectorPresence itemPresence)
    {
        logManager.WriteLog($"Library : Received connector operation for item {itemId}, the name of the item is {itemPresence.ItemName}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogManagement.LogLevel.Information);
        //Set requestor UI item in working state in the UI
        if(ci.CIType == EnumCIType.Application)
        {
            LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).IsWorking = true;
            LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).Status = LibraryItemState.UpdatingStatus;
            LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).StatusString = "Checking Presence";
        }

        ActionBlock<OperationType> actionBlock = new ActionBlock<OperationType>(async _operationType =>
        {
            logManager.WriteLog($"Library : Sending the operation to connector service : item {itemId}, the name of the item is {itemPresence.ItemName}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogManagement.LogLevel.Information);
            await connectorService.EnqueueConnectorOperations(operationType, ci, itemId, Settings.Default.ProjectLocalPath + @"\" + ci.ID.ToString(), ConnectorID, Settings.Default.DisplayLanguage, Settings.Default.ProjectLocalPath, itemPresence).ConfigureAwait(false);
        }, new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 1,
            CancellationToken = new CancellationTokenSource(TimeSpan.FromMinutes(5)).Token,
        });

        actionBlock.Post(operationType);
        actionBlock.Complete();
        actionBlock.Completion.Wait();
    }

然后这里命名为“connectorService”的服务就完成了它的工作。

在最后一行，如果我使用 actionBlock.Completion.Wait() 所有任务按顺序运行，我的 UI 被阻止。

如果我改用 await actionBlock.Completion()。 UI 没有被阻塞，而是并行运行。

因此，如果有人有建议，那就太好了!

更新:

我调整了 Jsteward 的 anwser 以满足我的需要:

我按照您的建议将 ActionBlock 声明为我的 View 模型的私有(private)成员。但是，当我按照您所说的那样消耗一个项目时，它是正确排队的操作，但是如果我扩展另一个项目，那么它的操作(也在他们的队列中)与第一个项目的操作并行运行。无论请求多少项目，这不是我希望一次只进行一项操作的行为。

所以我做了以下更改:
ActionBlock 在 vi​​ewmodel 的构造函数中被初始化一次:

public ViewModelCtor()
{
 actionBlock = new ActionBlock<ConnectorOperationArgWrapper>(async _connectorOperationArgWrapper =>
            {
                logManager.WriteLog($"Library : Sending the operation to connector service for {_connectorOperationArgWrapper.itemPresence.ItemName} on connector {connectorService.GetConnectorName(_connectorOperationArgWrapper.itemPresence.ConnectorId)}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogLevel.Information);
                LibraryItems.Single(l => l.CI.ID == _connectorOperationArgWrapper.ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == _connectorOperationArgWrapper.itemPresence.itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == _connectorOperationArgWrapper.itemPresence.ConnectorId).StatusString = "Cheking Presence";
                LibraryItems.Single(l => l.CI.ID == _connectorOperationArgWrapper.ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == _connectorOperationArgWrapper.itemPresence.itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == _connectorOperationArgWrapper.itemPresence.ConnectorId).Status = LibraryItemState.UpdatingStatus;
                await connectorService.EnqueueConnectorOperations(_connectorOperationArgWrapper.operationType, _connectorOperationArgWrapper.ci, _connectorOperationArgWrapper.itemPresence.itemId, Settings.Default.ProjectLocalPath + @"\" + _connectorOperationArgWrapper.ci.ID.ToString(), _connectorOperationArgWrapper.itemPresence.ConnectorId, Settings.Default.DisplayLanguage, Settings.Default.ProjectLocalPath, _connectorOperationArgWrapper.itemPresence).ConfigureAwait(false);
            }, new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 1,
            });
}

因此，被扩展的项目调用的方法现在看起来像这样:

public async Task EnqueueConnectorOperations(ConnectorOperationArgWrapper _args)
    {

        logManager.WriteLog($"Library : Received operation request for {_args.itemPresence.ItemName} on connector {connectorService.GetConnectorName(_args.itemPresence.ConnectorId)}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogLevel.Information);

        if (_args.ci.CIType == EnumCIType.Application)
        {
            LibraryItems.Single(l => l.CI.ID == _args.ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == _args.itemPresence.itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == _args.itemPresence.ConnectorId).IsWorking = true;
            LibraryItems.Single(l => l.CI.ID == _args.ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == _args.itemPresence.itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == _args.itemPresence.ConnectorId).Status = LibraryItemState.NeedsAttention;
            LibraryItems.Single(l => l.CI.ID == _args.ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == _args.itemPresence.itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == _args.itemPresence.ConnectorId).StatusString = "Waiting";
        }

        logManager.WriteLog($"Library : post actionblock", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogLevel.Information);
        await actionBlock.SendAsync(_args);

        //actionBlock.Complete();
        //await actionBlock.Completion;
    }

我用 actionBlock complete 和 completion 评论了该部分，因为我希望该 block 能够随时接收和排队请求，甚至每个项目可能多次。

到目前为止，它似乎有效，像我这样做是正确的还是我会面临一些麻烦？

Answer 1

现在你正在创建一个新的 ActionBlock对于每个操作。一个 ActionBlock有一个内部队列，您应该向其发送消息并让它使用单个 ActionBlock 顺序运行它们.通过重新排列事物并制作 ActionBlock一个类(class)成员，您将能够更好地控制它并等待每组 subview 项。

private ActionBlock<OperationType> actionBlock;

public void OnTreeViewExpand()
{
    //Re-initialize the actionblock for a new set of operations
    actionBlock = new ActionBlock<OperationType>(async _operationType =>
    {
        logManager.WriteLog($"Library : Sending the operation to connector service : item {itemId}, the name of the item is {itemPresence.ItemName}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogManagement.LogLevel.Information);
        await connectorService.EnqueueConnectorOperations(operationType, ci, itemId, Settings.Default.ProjectLocalPath + @"\" + ci.ID.ToString(), ConnectorID, Settings.Default.DisplayLanguage, Settings.Default.ProjectLocalPath, itemPresence).ConfigureAwait(false);
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 1,
        CancellationToken = new CancellationTokenSource(TimeSpan.FromMinutes(5)).Token,
    });
}

private async Task<bool> UpdateSubItemsStatus()
{
    foreach (var item in connectorsMenuItems)
    {
        await parent.Library.EnqueueConnectorOperations(Connectors.service.OperationType.CheckPresence, parent.CI, AssetId, item.ConnectorID, parent.ConnectorInterfaces.Single(c => c.ItemId == AssetId).ItemsConnectorPresence.Single(i => i.ConnectorId == item.ConnectorID));
    }

    //All items sent, signal completion
    actionBlock.Complete();
    await actionBlock.Completion;
    return true;
}

public Task EnqueueConnectorOperations(OperationType operationType, ConfigurationItem ci, Guid itemId, Guid ConnectorID, ItemConnectorPresence itemPresence)
{
    logManager.WriteLog($"Library : Received connector operation for item {itemId}, the name of the item is {itemPresence.ItemName}", System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(), LogManagement.LogLevel.Information);
    //Set requestor UI item in working state in the UI
    if (ci.CIType == EnumCIType.Application)
    {
        LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).IsWorking = true;
        LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).Status = LibraryItemState.UpdatingStatus;
        LibraryItems.Single(l => l.CI.ID == ci.ID).DeployableAssetMenuItems.Single(d => d.AssetId == itemId).ConnectorsMenuItems.Single(c => c.ConnectorID == ConnectorID).StatusString = "Checking Presence";
    }
    return actionBlock.SendAsync(operationType);
}