c# - 在 C# 中实现线程以进行快速、批量和连续读取的最佳实践？

Question

在 .NET 4.0 中应如何处理在 C# 中从设备读取批量数据？具体来说，我需要从 USB HID 设备中快速读取，该设备会发出超过 26 个数据包的报告，其中必须保留顺序。

我试过在 BackgroundWorker 线程中执行此操作。它一次从设备读取一个数据包，并在读取更多数据之前对其进行处理。这提供了相当不错的响应时间，但它很容易丢失数据包，并且单个数据包读取的开销成本加起来。

while (!( sender as BackgroundWorker ).CancellationPending) {
       //read a single packet
       //check for header or footer
       //process packet data
    }
}

在 C# 中读取此类设备的最佳做法是什么？

---

背景:

我的 USB HID 设备不断报告大量数据。数据分为 26 个数据包，我必须保留顺序。不幸的是，该设备只标记每个报告中的第一个和最后一个数据包，因此我需要能够捕获其间的所有其他数据包。

Answer 1

对于 .Net 4，您可以使用 BlockingCollection提供可供生产者和消费者使用的线程安全队列。 BlockingCollection.GetConsumingEnumerable()方法提供了一个枚举器，当使用 CompleteAdding() 将队列标记为已完成时，该枚举器会自动终止。并且是空的。

这是一些示例代码。在此示例中，有效负载是一个整数数组，但您当然可以使用所需的任何数据类型。

请注意，对于您的特定示例，您可以使用 the overload of GetConsumingEnumerable()它接受类型为 CancellationToken 的参数.

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    public static class Program
    {
        private static void Main()
        {
            var queue = new BlockingCollection<int[]>();

            Task.Factory.StartNew(() => produce(queue));

            consume(queue);

            Console.WriteLine("Finished.");
        }

        private static void consume(BlockingCollection<int[]> queue)
        {
            foreach (var item in queue.GetConsumingEnumerable())
            {
                Console.WriteLine("Consuming " + item[0]);
                Thread.Sleep(25);
            }
        }

        private static void produce(BlockingCollection<int[]> queue)
        {
            for (int i = 0; i < 1000; ++i)
            {
                Console.WriteLine("Producing " + i);
                var payload = new int[100];
                payload[0] = i;
                queue.Add(payload);
                Thread.Sleep(20);
            }

            queue.CompleteAdding();
        }
    }
}

对于 .Net 4.5 及更高版本，您可以使用 Microsoft's Task Parallel Library 中的高级类，它具有丰富的功能(乍一看可能有点令人生畏)。

下面是使用 TPL DataFlow 的相同示例:

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Demo
{
    public static class Program
    {
        private static void Main()
        {
            var queue = new BufferBlock<int[]>();

            Task.Factory.StartNew(() => produce(queue));
            consume(queue).Wait();

            Console.WriteLine("Finished.");
        }

        private static async Task consume(BufferBlock<int[]> queue)
        {
            while (await queue.OutputAvailableAsync())
            {
                var payload = await queue.ReceiveAsync();
                Console.WriteLine("Consuming " + payload[0]);
                await Task.Delay(25);
            }
        }

        private static void produce(BufferBlock<int[]> queue)
        {
            for (int i = 0; i < 1000; ++i)
            {
                Console.WriteLine("Producing " + i);
                var payload = new int[100];
                payload[0] = i;
                queue.Post(payload);
                Thread.Sleep(20);
            }

            queue.Complete();
        }
    }
}