异步任务、视频缓冲

我正在尝试理解 C# 中的任务，但仍然遇到一些问题。我正在尝试创建一个包含视频的应用程序。主要目的是从文件中读取视频（我使用 Emgu.CV）并通过 TCP/IP 发送它以在板上进行处理，然后以流（实时）方式返回。首先，我是连续做的。所以，读一Bitmap、从板上发送-接收以及绘图。但是读取位图并绘制它们需要太多时间。我想要一个传输、接收 FIFO 缓冲区来保存视频帧，以及一个不同的任务来完成发送和接收每个帧的工作。所以我想并行进行。我想我应该创建 3 个任务：

        tasks.Add(Task.Run(() => Video_load(video_path)));
        tasks.Add(Task.Run(() => Video_Send_Recv(video_path)));
        tasks.Add(Task.Run(() => VideoDisp_hw(32)));

我想“并行”运行。我应该使用什么类型的对象？并发队列？缓冲区块？或者只是一个列表？

感谢您的建议！我想问一件事。我正在尝试创建一个带有 2 个 TPL 块的简单控制台程序。 1 个块将是 Transform 块（获取消息，即“开始”）并将数据加载到列表中，另一个块将是 ActionBlock（仅从列表中读取数据并打印它们）。下面是代码：

namespace TPL_Dataflow
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Hello World!");
            Random randn = new Random();

            var loadData = new TransformBlock<string, List<int>>(async sample_string =>
           {
               List<int> input_data = new List<int>();
               int cnt = 0;

                if (sample_string == "start")
                {
                   Console.WriteLine("Inside loadData");
                   while (cnt < 16)
                   {
                       input_data.Add(randn.Next(1, 255));
                       await Task.Delay(1500);
                       Console.WriteLine("Cnt");
                       cnt++;
                   }
                                    }
                else
                {
                    Console.WriteLine("Not started yet");

                }
            return input_data;
           });


            var PrintData = new ActionBlock<List<int>>(async input_data =>
            {
                while(input_data.Count > 0)
                {


                    Console.WriteLine("output Data = " + input_data.First());
                    await Task.Delay(1000);
                    input_data.RemoveAt(0);
                    
                }
 

              });

            var linkOptions = new DataflowLinkOptions { PropagateCompletion = true };
            loadData.LinkTo(PrintData, input_data => input_data.Count() >0  );
            //loadData.LinkTo(PrintData, linkOptions);
            
            loadData.SendAsync("start");
            loadData.Complete();
            PrintData.Completion.Wait();

        }
    }
}

但它似乎以串行方式工作..我做错了什么？我尝试异步执行 while 循环。我想并行做这两件事。当从列表中获得数据时，然后绘制。

你可以使用TransformManyBlock<string, int> https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.transformmanyblock-2作为生产者区块，以及ActionBlock<int>作为消费者块。这TransformManyBlock将使用接受的构造函数进行实例化Func<string, IEnumerable<int>>代表，并通过了迭代器方法 https://learn.microsoft.com/en-us/dotnet/csharp/iterators (the Produce方法（下例中的方法）会一一生成值：

Random random = new Random();

var producer = new TransformManyBlock<string, int>(Produce);

IEnumerable<int> Produce(string message)
{
    if (message == "start")
    {
        int cnt = 0;
        while (cnt < 16)
        {
            int value;
            lock (random) value = random.Next(1, 255);
            Console.WriteLine($"Producing #{value}");
            yield return value;
            Thread.Sleep(1500);
            cnt++;
        }
    }
    else
    {
        yield break;
    }
}

var consumer = new ActionBlock<int>(async value =>
{
    Console.WriteLine($"Received: {value}");
    await Task.Delay(1000);
});

producer.LinkTo(consumer, new() { PropagateCompletion = true });

producer.Post("start");
producer.Complete();
consumer.Completion.Wait();

不幸的是，生产者必须在产生每个值之间的空闲期间阻塞工作线程（Thread.Sleep(1500);），因为TransformManyBlock目前没有接受的构造函数Func<string, IAsyncEnumerable<int>>。这可能会在下一版本中修复TPL数据流 https://learn.microsoft.com/en-us/dotnet/standard/parallel-programming/dataflow-task-parallel-library图书馆。你可以追踪this https://github.com/dotnet/runtime/issues/30863GitHub 问题，了解此功能何时发布。

替代解决方案：您可以保持它们不链接，然后手动将生产者生成的值发送给消费者，而不是显式链接生产者和消费者。在这种情况下，两个块都是ActionBlock https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.actionblock-1s:

Random random = new Random();

var consumer = new ActionBlock<int>(async value =>
{
    Console.WriteLine($"Received: {value}");
    await Task.Delay(1000);
});

var producer = new ActionBlock<string>(async message =>
{
    if (message == "start")
    {
        int cnt = 0;
        while (cnt < 16)
        {
            int value;
            lock (random) value = random.Next(1, 255);
            Console.WriteLine($"Producing #{value}");
            var accepted = await consumer.SendAsync(value);
            if (!accepted) break; // The consumer has failed
            await Task.Delay(1500);
            cnt++;
        }
    }
});

PropagateCompletion(producer, consumer);

producer.Post("start");
producer.Complete();
consumer.Completion.Wait();

async void PropagateCompletion(IDataflowBlock source, IDataflowBlock target)
{
    try { await source.Completion.ConfigureAwait(false); } catch { }
    var ex = source.Completion.IsFaulted ? source.Completion.Exception : null;
    if (ex != null) target.Fault(ex); else target.Complete();
}

这种方法的主要困难是如何将生产者的完成情况传播给消费者，以便最终两个块都完成。显然你不能使用new DataflowLinkOptions { PropagateCompletion = true }配置，因为块没有显式链接。你也不能Complete手动消费者，因为在这种情况下，它会过早地停止接受来自生产者的值。这个问题的解决方案是PropagateCompletion方法如上例所示。