c# - 如何异步迭代包含分层元素的 ObservableCollection？

Question

我使用 MSVS 2015 编写 C# WPF 应用程序。我对分层 ObservableCollection 的异步处理很感兴趣。我有一个包含分层元素的 ObservableCollection 实例:

public class Group : ProfileElementType
{
    . . . . . .
    const ushort deviceAddress = 1;
    . . . . . .
    private ObservableCollection<ProfileElementType> _childProfileElenents;
    [Browsable(false)]
    public ObservableCollection<ProfileElementType> ChildProfileElenents
    {
        get { return this._childProfileElenents; }
        set { this_childProfileElenents = value; }
    }
    public Group()
    {
        . . . . .
        this.ChildProfileElenents = new ObservableCollection<ProfileElementType>();
        . . . . .
    }
    . . . . .
}

ProfileElementType 类是Group 类和Register 类的基础类。您可以在我在 Why is InvalideOperationException thrown when I try to serialize to XML an ObservableCollection containing hierarchical elements? 上的帖子中查看 ProfileElementType、Group 和 Register 类的详细定义。我的集合可以包含 Register 实例和 Group 实例。每个 Group 实例都可以在其 ChildProfileElenents 中涉及其他 Group 实例和 Register 实例可观察的集合。我有一个根组实例，它具有所有其他组并在其 ChildProfileElenents 集合(根集合)中注册。我编写了一个同步处理根集合的递归函数。在这里:

private void pollDeviceRegisters(ObservableCollection<ProfileElementType> collection)
{
    if (collection == null)
       return;
    foreach (ProfileElementType elem in collection)
    {
        if (elem.ElementTypeName == "Group")
        {
            Group group = elem as Group;
            pollDeviceRegisters(group.ChildProfileElenents);
        }
        else
        {
            Register reg = elem as Register;
            // Get this register' value from outer device via serial port using MODBUS RTU protocol.
            ushort[] aRes = ReadHoldingRegisters(deviceAddress, reg.Number, 1);
            reg.CurrentValue = aRes[0].ToString("X");
        }
    }
}

外部设备中有很多寄存器 - 不少于2000。所以这个synchrone sequential function由于每次迭代中的类型转换和串口读取超时的原因而运行缓慢。所以我的应用程序在函数工作时挂起。如果你告诉我如何异步编写上述操作，我将不胜感激。例如，如何使用 TPL 中的 Parallel.ForEach 方法或 async/await 方法编写 pollDeviceRegisters 递归函数。请帮忙。

Answer 1

您可以使用 BlockingCollection 读取串行端口，如果这是瓶颈，则将该部分移至新线程。

private Task pollDeviceRegisters(ICollection<ProfileElementType> collection)
{
    var blockingCollection = new BlockingCollection<Register>();

    // This starts three threads to start reading the serial ports (arbitrary number of threads to start you will need to see what is optimal)
    var pollingTask = Task.WhenAll(ReadSerialPort(blockingCollection),
        ReadSerialPort(blockingCollection),
        ReadSerialPort(blockingCollection))
        .ContinueWith(task =>
        {
            blockingCollection.Dispose();
            return task;
        });

    BuildDeviceRegisters(collection, blockingCollection);

    return pollingTask;
}

// Creates a new thread and waits on the blocking collection
private Task ReadSerialPort(BlockingCollection<Register> collection)
{
    return Task.Run(() =>
    {
        foreach (var reg in collection.GetConsumingEnumerable())
        {
            // Get this register' value from outer device via serial port using MODBUS RTU protocol.
            var aRes = ReadHoldingRegisters(deviceAddress, reg.Number, 1);
            reg.CurrentValue = aRes[0].ToString("X");
        }
    });
}

// flattens the hierarchical data
private static void BuildDeviceRegisters(ICollection<ProfileElementType> collection,
    BlockingCollection<Register> blockingCollection)
{
    if (collection != null)
    {
        //using a stack instead of recursion (could switch to ConcurrentStack and use Parallel.ForEach() but wouldn't mess with it if not bottle neck)
        var stack = new Stack<ICollection<ProfileElementType>>(new[] {collection});
        while (stack.Count > 0)
        {
            var element = stack.Pop();
            foreach (var item in element)
            {
                var group = item as Group;
                if (group != null)
                {
                    // push the tree branch on to the stack
                    stack.Push(group.ChildProfileElenents);
                }
                else
                {
                    blockingCollection.Add((Register)item);
                }
            }
        }
    }

    // Mark we are done adding to the collection
    blockingCollection.CompleteAdding();
}

我没有测试这段代码。

此外，这还会返回一个任务，需要等待该任务才能知道串行端口的读取何时完成。一旦您开始使用 TPL，您将看到许多方法需要更改为异步和等待。如果您向集合中添加太多内容并占用大量内存，您还可以限制添加到阻塞集合中，这将使添加方法阻塞，直到消费者从集合中获取。