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

Question

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

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

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

这是我所做的：

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

我想要的是，每次我展开一个项目时，树视图中的所有子项目都会通过服务检查它们在 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 声明为我的视图模型的私有成员。但是，当我按照您所说的那样使用一个项目时，它是正确排队的操作，但是如果我扩展另一个项目，那么它的操作（也在他们的队列中）与第一个项目的操作并行运行。无论请求多少项目，我都不希望一次只进行一次操作。

所以我做了以下更改： 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 评论了该部分，因为我希望该块能够随时接收和排队请求，甚至每个项目可能多次。

到目前为止，它似乎奏效了，我这样做是对的还是我会面临一些麻烦？

Answer 1

现在您正在为每个操作创建一个新的ActionBlock。 ActionBlock 有一个内部队列，您应该向其发送消息并让它使用单个 ActionBlock 按顺序运行它们。通过重新排列事物并使ActionBlock 成为班级成员，您将能够更好地控制它并等待每组子视图项。

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);
}

Answer 2

将BlockingCollection 与TaskCompletionSource 一起使用

    using System;
    using System.Collections;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Linq;
    using System.Reactive;
    using System.Reactive.Linq;
    using System.Reactive.Subjects;
    using System.Threading.Tasks;

    namespace test
    {
        class Program
        {
            class WorkItem
            {
                public int Id { get; set; } // you can make it store more things
                public TaskCompletionSource<DateTime> TaskSource { get; } = new TaskCompletionSource<DateTime>();
            }

            class Worker : IDisposable
            {
                private BlockingCollection<WorkItem> _queue;
                private Task _consumer;

                public Worker()
                {
                    _queue = new BlockingCollection<WorkItem>();

                    _consumer = Task.Run(async () =>
                    {
                        foreach (var item in _queue.GetConsumingEnumerable())
                        {
                            await Task.Delay(1000); // some hard work
                            item.TaskSource.TrySetResult(DateTime.Now); // try is safer
// you can return whatever you want
                        }
                    });
                }

                public Task<DateTime> DoWork(int i) // return whatever you want
                {
                    var workItem = new WorkItem { Id = i };
                    _queue.Add(workItem);

                    return workItem.TaskSource.Task;
                }
                public void Dispose()
                {
                    _queue.CompleteAdding();
                }
            }

            public static void Main(string[] args)
            {
                using (var worker = new Worker())
                {
                    Task.Run(async () =>
                    {
                        var tasks = Enumerable.Range(0,10).Select(x => worker.DoWork(x)).ToArray();

                        var time = await tasks[1];
                        Console.WriteLine("2nd task finished at " + time);

                        foreach (var task in tasks)
                        {
                            time = await task;
                            Console.WriteLine("Task finished at " + time);
                        }

                        Console.ReadLine();
                    }).Wait();
                }



            }
        }
    }
    // output
    2nd task finished at 2019-01-22 19:14:57
    Task finished at 2019-01-22 19:14:56
    Task finished at 2019-01-22 19:14:57
    Task finished at 2019-01-22 19:14:58
    Task finished at 2019-01-22 19:14:59
    Task finished at 2019-01-22 19:15:00
    Task finished at 2019-01-22 19:15:01
    Task finished at 2019-01-22 19:15:02
    Task finished at 2019-01-22 19:15:03
    Task finished at 2019-01-22 19:15:04
    Task finished at 2019-01-22 19:15:05

这使您可以轻松地等待命令中的单个项目，例如，不会阻塞 UI 线程。